Server-Push-Benachrichtigungen empfangen

Der enaio-Server kann Clients aktiv benachrichtigen. Dazu öffnet der Client neben der normalen Job-Verbindung einen zweiten TCP-Kanal mit vertauschten Rollen: Nach der Registrierung sendet nur noch der Server Job-Calls (die Notifications), und der Client beantwortet jeden davon mit einer Job-Response.

Empfohlen wird die hochsprachige, typisierte API ecm.notification (siehe unten). Wer volle Kontrolle braucht, nutzt die Low-Level-Funktionen in ecmind_blue_client.rpc.sync_api bzw. ecmind_blue_client.rpc.async_api (weiter unten).

1. Hochsprachig: ecm.notification (empfohlen)

ecm.notification öffnet die Callback-Kanäle selbst - pro konfiguriertem Server bindet es einen geteilten Callback-Kanal an die geteilte Server-Session des Pools (kein eigener Notification-Login; der Kanal hält die Session offen) - und liefert typisierte Ereignisse. Es gibt drei Opener; jeder gibt eine NotificationSubscription zurück (mit stop() bzw. als Context-Manager):

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

Überwacht Serverjob-Aufrufe (krn.RegisterJobCall) und liefert typisierte RegisteredJobCall-Ereignisse. jobs=None überwacht alle Jobs (Firehose).

ecm.notification.messages(handler)

Direkte Client-Nachrichten (krn.SendMessageToClients → adm.<Message>, z. B. adm.Message, adm.Ping, adm.Shutdown) als typisierte ClientMessage.

ecm.notification.listen(handler)

Jede rohe Notification (generisch, u. a. asynchrone wfm-Pushes wie 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"])
# ... läuft im Hintergrund ...
sub.stop()

# oder als Context-Manager:
with ecm.notification.messages(lambda m: print(m.message, m.info, m.text)):
    ...

Async funktioniert identisch mit await und async def-Handlern:

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

Mehrere Server: Standardmäßig (servers=None) wird auf allen konfigurierten Servern registriert - Notifications sind serverspezifisch. NotificationSubscription.server_count zeigt, auf wie vielen Servern der Listener aktiv wurde; nicht erreichbare Server werden übersprungen.

Mehrere Listener / Stoppen: Listener auf demselben Server teilen sich Kanal und eine zentral verwaltete Registrierung. Einen Listener zu stoppen lässt die anderen weiterlaufen. Ein abgebrochener Kanal wird automatisch neu verbunden.

job_calls ist ein Administrations-/Monitoring-Feature: Die Ereignisse enthalten die Ein-/Ausgabeparameter der überwachten Jobs unmaskiert - inklusive z. B. des verschlüsselten Passworts bei krn.SessionLogin. Nicht an nicht vertrauenswürdige Konsumenten weitergeben.

Zur Korrelation von RegisteredJobCall-Ereignissen: Es gibt kein Server-seitiges Korrelations-ID. before_too=True liefert vor und nach dem Aufruf ein Ereignis (phase), aber Before/After lassen sich nur heuristisch paaren (Reihenfolge + job/instance/connect_ip/user + time).

2. Kommandozeile: ecm-callback-listen

Zum Mitlesen des Push-Streams aus der Shell installiert das Paket das Konsolen-Skript ecm-callback-listen. Es lauscht auf allen im --servers-String angegebenen Servern, öffnet je Server den Callback-Kanal und gibt jede Notification roh aus.

# Passiv: nur direkte Nachrichten (adm.<Message>)
ecm-callback-listen --servers localhost:4000:1 --username root --password optimal

# Mehrere Server (lauscht auf allen)
ecm-callback-listen --servers "srv1:4000:1#srv2:4000:1" -u root -p optimal

# Bestimmte Job-Aufrufe überwachen
ecm-callback-listen --servers localhost:4000:1 -u root -p optimal --jobs dms.XMLInsert,dms.UpdateDocumentData

# ALLE Job-Aufrufe (Firehose; Job-Parameter inkl. Passwörter im Klartext-Feld)
ecm-callback-listen --servers localhost:4000:1 -u root -p optimal --all-jobs

Der --servers-String hat das Pool-Format hostname:port:weight, mehrere durch # getrennt. Weitere Optionen: --before (auch Before-Event), --errors-only, --no-ssl, --name. Ohne --password wird interaktiv gefragt. Beenden mit Strg+C.

3. Low-Level: Callback-Kanal direkt

Wer den Kanal selbst steuern will, nutzt die Low-Level-Funktionen.

4. Funktionsprinzip

  1. Auf der normalen Session-Verbindung die Kanal-GUID holen: Job krn.GetChannelGUID liefert die GUID des Kommunikationskanals der Session.

  2. Mit sync_open_callback() (bzw. async_open_callback()) eine dedizierte Verbindung öffnen. Sie registriert sich beim Server mit dieser GUID und empfängt fortan die Notifications, die für diese Session anfallen.

  3. In einer Schleife sync_callback_next() (bzw. await async_callback_next()) aufrufen: Die Funktion wartet auf die nächste Notification, ruft den übergebenen Handler auf und bestätigt die Notification beim Server.

Welche Notifications der Server zustellt, wird auf der Session-Verbindung gesteuert, z. B. über krn.AppsEventsSubscribe (Serverereignisse) oder abn.Add mit Channel=0 (Objekt-Abonnements über den internen Kanal). Die Zustellung hängt von der Serverkonfiguration ab.

Direkt adressieren lässt sich ein Client mit krn.SendMessageToClients: Eine an Computer/Instance/User der Empfänger-Session gerichtete Nachricht wird als Job-Call adm.<Message> mit den Parametern Info und Text über den Callback-Kanal zugestellt (verifiziert gegen enaio 12.0).

5. Minimal-Beispiel (synchron)

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. Kanal-GUID der Session-Verbindung holen
guid = ecm.execute(Jobs.KRN_GETCHANNELGUID, Flags=0).get("ChannelGUID", str)

# 2. Callback-Kanal registrieren
connection = sync_open_callback("localhost", guid)


# 3. Notifications verarbeiten
def handler(notification):
    print(notification.name, [(p.name, p.value) for p in notification.parameters])
    return None  # None = Standard-Acknowledgement (Return-Code 0)


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

sync_callback_next() blockiert, bis eine Notification eintrifft; typischerweise läuft die Schleife in einem eigenen Thread. Das optionale timeout (Sekunden) begrenzt nur das Warten auf den Anfang einer Notification: Bei Ablauf wird TimeoutError geworfen und der Kanal bleibt nutzbar.

6. Minimal-Beispiel (asynchron)

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)

Zum Beenden einer wartenden Empfangsschleife connection.close() aufrufen (auch aus einem anderen Thread bzw. Task). Nach einem Abbruch mitten im Lesen darf der Kanal nicht weiterverwendet werden.

7. Der Handler

Der Handler erhält eine Notification (name, parameters, internal_parameters, files, Zugriffshelfer get()) und bestimmt die Antwort an den Server:

return None

Standard-Acknowledgement mit Return-Code 0.

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

Eigene Antwort mit Return-Code, optionalen Parametern und Fehlern.

Ausnahme im Handler

Wird geloggt; es wird das Standard-Acknowledgement gesendet und der Kanal bleibt nutzbar.

8. Hinweise

  • Kein automatischer Reconnect: Wird die Verbindung geschlossen oder bricht sie ab, muss ein neuer Kanal geöffnet werden (ConnectionError). Schlägt der Empfang bereits vor der ersten Notification fehl, ist meist die Kanal-GUID falsch; die Fehlermeldung weist darauf hin.

  • Protokollversion: Der Kanal nutzt Protokollversion 50 - jede Notification und jede Antwort trägt eine SHA-1-Prüfsumme, die beim Empfang validiert wird.

  • Dateien: Notifications können Dateien enthalten; sie landen wie bei Job-Responses als JobResponseFile in notification.files (kleine Dateien im Speicher, große als Temp-Datei, Schwelle file_cache_byte_limit).