Subscribing Asynchronously

Overview

Callback subscriptions push frames to you. Async subscriptions let you pull them. subscribe() returns a subscription handle, await sub.recv_async() takes one frame at a time, and six streams each run their own coroutine driven concurrently by asyncio.gather. Compared with callbacks, async gives you control over pacing: the code reads sequentially and can await other I/O directly. The cost is that the subscription queue fills up while you’re not reading, and silently drops frames once it’s full. This page measures exactly where that boundary sits. Matching example: examples/python/wuji_glove/1.subscribe_async.py. For field definitions of each stream, see SDK Reference. This page covers only how to run it and what the measurements show.

The C SDK glove examples offer callback subscriptions only, with no async counterpart. This page covers Python.

Prerequisites

Hardware

Item	Requirement
Wuji Glove	1
Connection	Ethernet
Network	The device uses a static IP. Configure your network adapter on the same subnet. The device verified here listens on port 50001

This example only reads. It writes no parameters.

Software

Item	Requirement
Operating system	Ubuntu 22.04 LTS or later
Wuji SDK	v2026.8.31 or later
Python	3.10 or later

pip install wuji-sdk

Before You Start

Read the callback page first. Frame structures for all six streams, the -1.0 invalid-channel marker, and how the numbers 744, 572, and 526 relate all live there, and this page doesn’t repeat them.

Data Flow

The queues live inside the SDK, filled by firmware pushes and decoupled from your consuming coroutines. That decoupling is where the frame-loss risk comes from.

Queue Measurements

The Queue Holds 1024 Frames

The queue fills from the moment a subscription opens, even if you never read. Measured on the tactile stream at 132 Hz:

Idle time after subscribing	Frames produced	How far behind the first `recv_async()` frame lands
2 seconds	about 264	300 frames
5 seconds	about 655	654 frames
15 seconds	about 1965	1023 frames

In the first two rows, the lag equals the production count, so nothing was lost. The third row stops at 1023, meaning everything beyond that was already discarded. The queue holds 1024 frames, matching the SDK’s internal broadcast channel setting.

Drops Take the Oldest Frames, Never a Sample Here and There

Across all three runs, reading 400 consecutive frames after the first one gives a header.seq step set of exactly [1]. On overflow the queue discards the oldest batch at its head. What remains stays contiguous, with no holes punched in the middle.

Drops Raise No Exception, Only a WARN Log

After a 12-second backlog, reading resumes and the SDK prints:

[<timestamp> WARN  wuji_sdk] Subscription for 'tactile' lagged 549 messages

recv_async() returns the next frame normally, raises nothing, and carries no drop marker in its return value. Your application code has no visibility into those 549 lost frames. The log is the only signal. For work that depends on data completeness, set the SDK log level to WARN or higher and route it into your own logging system rather than letting it scroll past in a terminal.

Fast Consumers Lose Nothing

Keep up and you lose nothing. Reading 200 frames with consecutive await recv_async() calls takes 1.52 seconds, an effective 131 Hz that matches the stream’s own rate, with every seq step equal to 1.

Runtime Handling

Branch point	Behavior	What to do
Consumer slower than producer	The queue grows and latency climbs until it caps at 1024 frames	Lower the stream rate, or do only a minimal copy inside a callback subscription
Queue overflow	Oldest frames drop silently, with only a WARN log	Watch the SDK log, or switch to `subscribe_with_callback`
Blocking call inside a coroutine	The whole event loop stalls and all six streams back up together	Move blocking work to `run_in_executor`
Device already in use	Connecting raises `Session already exists on device`	Wait for the previous session’s heartbeat to time out, then retry
Device disconnects	`recv_async()` raises `Disconnected`	Catch it, reconnect, and resubscribe
Several gloves on the network	The example connects to each glove it finds	Pass `--sn` to pick one

The example’s while True loop has no exit condition and stops on KeyboardInterrupt. Its finally block only calls manager.disconnect_all(), leaving subscription handles to be reclaimed at process exit. A long-running service needs its own sub.close().

Integration Steps

import asyncio
from wuji_sdk import SdkManager, DeviceType, WujiGlove, TactileFrame, HandSkeleton

async def consume_tactile(glove: WujiGlove):
    sub = glove.tactile().subscribe()
    try:
        while True:
            frame: TactileFrame = await sub.recv_async()
            # -1.0 marks an invalid channel. Filter before computing statistics
            valid = [v for v in frame.data if v >= 0.0]
            print(f"[{glove.device_name}][Tactile] seq={frame.header.seq} "
                  f"max={max(valid):.3f}")
    finally:
        sub.close()

async def consume_skeleton(glove: WujiGlove):
    sub = glove.hand_skeleton().subscribe()
    try:
        while True:
            skeleton: HandSkeleton = await sub.recv_async()
            tip = next(j for j in skeleton.joints if j.name == "index_finger_tip")
            p = tip.pose.position
            print(f"[{glove.device_name}][Skeleton] "
                  f"index_tip=[{p[0]:+.3f}, {p[1]:+.3f}, {p[2]:+.3f}]")
    finally:
        sub.close()

async def main():
    manager = SdkManager.instance()

    # 1. Discover devices and filter by type
    devices = [d for d in manager.scan() if d.device_type == DeviceType.WujiGlove]

    # 2. Connect. With several devices around, name an sn instead of iterating
    glove = manager.connect(sn=devices[0].sn, device_name="glove_0")
    print(f"Connected: {glove.serial_number} "
          f"(FW={glove.version().get()}, {glove.hand_side().get()})")

    # 3. One coroutine per stream, driven concurrently by gather
    try:
        await asyncio.gather(
            consume_tactile(glove),
            consume_skeleton(glove),
        )
    finally:
        manager.disconnect_all()

asyncio.run(main())

subscribe() is a synchronous call, and the queue starts filling the moment it returns. The longer the gap between connecting and your first recv_async(), the older that first frame is. When you need data starting now, move subscribe() as close to the consuming loop as you can.

Running It

python examples/python/wuji_glove/1.subscribe_async.py --sn WG1KA00XXXXXXXXX

Found 1 device(s)
  SN=WG1KA00XXXXXXXXX, Type=DeviceType.WujiGlove, Address=192.168.1.100:50001
Connected: WG1KA00XXXXXXXXX (FW=0.11.2, right)
Subscribed to 6 streams. Ctrl+C to stop.

[glove_0][EmfPoses] thumb=[+0.067, +0.000, -0.059]
[glove_0][Tactile] max=0.3
[glove_0][JointAngles] conf=['0.47', '0.41', '0.74', '0.71', '0.47']
[glove_0][Skeleton] wrist=[+0.000, +0.000, +0.000] joints=21
[glove_0][Zones] palm=-0.1 thumb=-0.1
[glove_0][PointCloud] points=526 frame=r_wrist

Measured on a right-hand Wuji Glove resting on a desk, unworn. Five seconds produced 3031 lines, with per-stream frame counts of 506, 506, 505, 502, 502, and 501. No two streams differ by more than 5 frames, so the event loop favors none of them and gather schedules fairly. The palm=-0.1 on the Zones line comes from the example’s avg() including -1.0 invalid channels in its mean, not from the glove sensing a reversed force. The callback page explains the cause and the fix.

Adapting It

Callback or Async

	Callback	Async
Frame direction	The SDK pushes to you	You pull
Thread	An SDK worker thread per subscription	Your event loop, single-threaded
When you fall behind	Frames back up in the queue	Frames back up in the queue
Fits	Low-latency forwarding, writing to a queue	Logic that cooperates with other `await` calls

Queue semantics are identical, and the 1024-frame boundary applies to callbacks too. Callbacks keep your event loop free. Async composes naturally with coroutines such as HTTP requests and database writes.

Don’t Block Inside a Coroutine

The event loop is single-threaded. Calling time.sleep(), writing files synchronously, or making blocking HTTP requests inside a coroutine backs up all six streams at once. Hand blocking work to a thread pool:

loop = asyncio.get_running_loop()
await loop.run_in_executor(None, write_frame_to_disk, frame)

All six streams open means about 640 recv_async() wakeups per second at rest. For tactile analysis, keep tactile or tactile_point_cloud alone and event-loop pressure drops immediately.

Lower the Stream Rate

130 Hz is higher than most applications need. Subscription handles support rate control, described in Adjusting Stream Rate. At a lower rate, those 1024 queued frames cover a longer window, which also buys more tolerance for jitter.

Subscribing With Callbacks: frame structures for all six streams and invalid-channel handling
Recording Data: write these streams to a replayable file
Adjusting Stream Rate: bring 130 Hz down to what your application needs
Data Subscription: subscription model and lifecycle
Common Data Types: FrameHeader, TactileFrame, HandSkeleton, and related structures

​Overview

​Prerequisites

​Hardware

​Software

​Before You Start

​Data Flow

​Queue Measurements

​The Queue Holds 1024 Frames

​Drops Take the Oldest Frames, Never a Sample Here and There

​Drops Raise No Exception, Only a WARN Log

​Fast Consumers Lose Nothing

​Runtime Handling

​Integration Steps

​Running It

​Adapting It

​Callback or Async

​Don’t Block Inside a Coroutine

​Subscribe Only to What You Need

​Lower the Stream Rate

​Related

Overview

Prerequisites

Hardware

Software

Before You Start

Data Flow

Queue Measurements

The Queue Holds 1024 Frames

Drops Take the Oldest Frames, Never a Sample Here and There

Drops Raise No Exception, Only a WARN Log

Fast Consumers Lose Nothing

Runtime Handling

Integration Steps

Running It

Adapting It

Callback or Async

Don’t Block Inside a Coroutine

Subscribe Only to What You Need

Lower the Stream Rate

Related