Receiving server push notifications

The enaio server can actively notify clients. To receive these notifications the client opens a second TCP channel next to the regular job connection, with reversed roles: after registration only the server sends job calls (the notifications), and the client answers each of them with a job response.

The recommended entry point is the high-level, typed ecm.notification API (below). For full control, use the low-level functions in ecmind_blue_client.rpc.sync_api / ecmind_blue_client.rpc.async_api (further down).

ecm.notification opens the callback channels itself — per configured server it binds one shared callback channel to the pool’s shared server session (no separate notification login; the channel keeps that session alive) — and delivers typed events. Three openers, each returning a NotificationSubscription (with stop() or as a context manager):

ecm.notification.job_calls(handler, jobs=…​, before_too=False, errors_only=False, files_too=False)

Monitors server job calls (krn.RegisterJobCall) and dispatches typed RegisteredJobCall events. jobs=None monitors all jobs (firehose).

ecm.notification.messages(handler)

Direct client messages (krn.SendMessageToClients → adm.<Message>, e.g. adm.Message, adm.Ping, adm.Shutdown) as typed ClientMessage.

ecm.notification.listen(handler)

Every raw Notification (generic, including asynchronous wfm pushes such as inbox updates).

 
from ecmind_blue_client.ecm import ECM
from ecmind_blue_client.pool import SyncPoolClient

ecm = ECM(SyncPoolClient(servers="localhost:4000:1", username="<user>", password="<pass>"))

def on_call(ev):  # ev: RegisteredJobCall
    print(ev.job, ev.phase.name, ev.user, ev.params_in)

sub = ecm.notification.job_calls(on_call, jobs=["dms.XMLInsert", "dms.UpdateDocumentData"])
# ... runs in the background ...
sub.stop()

# or as a context manager:
with ecm.notification.messages(lambda m: print(m.message, m.info, m.text)):
    ...

Async works the same with await and async def handlers:

sub = await ecm.notification.job_calls(on_call, jobs=["dms.XMLInsert"])
await sub.stop()

Multiple servers: by default (servers=None) the listener registers on all configured servers — notifications are server-specific. NotificationSubscription.server_count reports how many servers it became active on; unreachable servers are skipped.

Multiple listeners / stopping: listeners on the same server share the channel and one centrally managed registration. Stopping one listener leaves the others running. A dropped channel reconnects automatically.

Stopping under cancellation: stop() frees the channel even when the calling task is cancelled. That is the case when a client disconnects from a streaming response under Starlette/FastAPI, where anyio re-delivers the cancellation on every loop iteration and every await in the cleanup raises immediately: the channel and its receiver are torn down before the first await, dropping the job-call registration continues shielded, and a listener started later opens a fresh channel on the same server. If a hub loses its receiver because that task was cancelled from outside, the next opener closes the orphaned channel and reports it as ECMWrongStateException instead of silently delivering no events; a second call opens a channel again.

job_calls is an administrative/monitoring feature: the events carry the monitored jobs' input/output parameters verbatim — including, for example, the encrypted password of krn.SessionLogin calls. Do not forward it to untrusted consumers.

On correlating RegisteredJobCall events: there is no server-side correlation id. before_too=True delivers an event before and after the call (phase), but before/after can only be paired heuristically (delivery order + job/instance/connect_ip/user + time).

2. Command line: ecm-callback-listen

To watch the push stream from the shell, the package installs the console script ecm-callback-listen. It listens on all servers given in the --servers string, opens the callback channel per server and prints every notification raw.

# Passive: only direct messages (adm.<Message>)
ecm-callback-listen --servers localhost:4000:1 --username root --password optimal

# Multiple servers (listens on all of them)
ecm-callback-listen --servers "srv1:4000:1#srv2:4000:1" -u root -p optimal

# Monitor specific job calls
ecm-callback-listen --servers localhost:4000:1 -u root -p optimal --jobs dms.XMLInsert,dms.UpdateDocumentData

# ALL job calls (firehose; job parameters incl. passwords in the clear-text field)
ecm-callback-listen --servers localhost:4000:1 -u root -p optimal --all-jobs

The --servers string uses the pool format hostname:port:weight, multiple separated by #. Further options: --before (also before-call events), --errors-only, --no-ssl, --name. Without --password it prompts interactively. Stop with Ctrl+C.

3. Low-level: the callback channel directly

For full control over the channel, use the low-level functions.

4. How it works

  1. Fetch the channel GUID on the regular session connection: job krn.GetChannelGUID returns the GUID of the session’s communication channel.

  2. Open a dedicated connection with sync_open_callback() (or async_open_callback()). It registers itself at the server with that GUID and from then on receives the notifications generated for that session.

  3. Call sync_callback_next() (or await async_callback_next()) in a loop: the function waits for the next notification, invokes the given handler and acknowledges the notification to the server.

Which notifications the server delivers is controlled on the session connection, e.g. via krn.AppsEventsSubscribe (server events) or abn.Add with Channel=0 (object subscriptions over the internal channel). Delivery depends on the server configuration.

A client can be addressed directly with krn.SendMessageToClients: a message targeted at the receiver session’s Computer/Instance/User is delivered over the callback channel as a job call named adm.<Message> with the parameters Info and Text (verified against enaio 12.0).

5. Minimal example (synchronous)

from ecmind_blue_client.ecm import ECM
from ecmind_blue_client.pool import SyncPoolClient
from ecmind_blue_client.rpc import Jobs
from ecmind_blue_client.rpc.sync_api import sync_callback_next, sync_open_callback

ecm = ECM(SyncPoolClient(servers="localhost:4000:1", username="<user>", password="<pass>"))

# 1. Fetch the channel GUID of the session connection
guid = ecm.execute(Jobs.KRN_GETCHANNELGUID, Flags=0).get("ChannelGUID", str)

# 2. Register the callback channel
connection = sync_open_callback("localhost", guid)


# 3. Process notifications
def handler(notification):
    print(notification.name, [(p.name, p.value) for p in notification.parameters])
    return None  # None = default acknowledgement (return code 0)


while True:
    sync_callback_next(connection, handler, timeout=30)

sync_callback_next() blocks until a notification arrives; typically the loop runs in a dedicated thread. The optional timeout (seconds) only limits waiting for the start of a notification: on expiry a TimeoutError is raised and the channel remains usable.

6. Minimal example (asynchronous)

from ecmind_blue_client.rpc.async_api import async_callback_next, async_open_callback

connection = await async_open_callback("localhost", guid)


async def handler(notification):
    print(notification.name)
    return None


while True:
    await async_callback_next(connection, handler)

To stop a waiting receive loop, call connection.close() (also from another thread or task). After an abort in the middle of a read the channel must not be reused.

7. The handler

The handler receives a Notification (name, parameters, internal_parameters, files, access helper get()) and determines the answer sent to the server:

return None

Default acknowledgement with return code 0.

return CallbackResult(return_code=…​, parameters=…​, errors=…​)

Custom answer with return code, optional parameters and errors.

Exception in the handler

Logged; the default acknowledgement is sent and the channel stays usable.

8. Notes

  • No automatic reconnect: when the connection is closed or fails, a new channel has to be opened (ConnectionError). If receiving fails before the first notification, the channel GUID is usually wrong; the error message points that out.

  • Protocol version: the channel uses protocol version 50 - every notification and response carries a SHA-1 checksum, which is validated on receipt.

  • Files: notifications can carry files; as with job responses they end up as JobResponseFile in notification.files (small files in memory, large ones as a temp file, threshold file_cache_byte_limit).