Pourquoi le client python ne reçoit-il pas les événements SSE ?

Question

J'ai un client python qui écoute les événements SSE d'un serveur avec une API node.js.

Le flux est que j'ai envoyé un événement à l'API node.js via call_notification.py et exécuter seevents.py en boucle en utilisant run.sh (voir ci-dessous)

Cependant, je ne vois pas que le client python reçoive cet événement SSE. Avez-vous une idée de la raison de cette situation ?

appel_notification.py

import requests
input_json = {'BATS':'678910','root_version':'12A12'}
url = 'http://company.com/api/root_event_notification?params=%s'%input_json
response = requests.get(url)
print response.text

API node.js

app.get("/api/root_event_notification", (req, res, next) => {
  console.log(req.query.params)
  var events = require('events');
  var eventEmitter = new events.EventEmitter();

  //Create an event handler:
  var myEventHandler = function () {
    console.log('new_root_announced!');
    res.status(200).json({
          message: "New root build released!",
          posts: req.query.params
        });
  }

seevents.py (client python écoutant les événements SSE)

import json
import pprint
import sseclient

def with_urllib3(url):
    """Get a streaming response for the given event feed using urllib3."""
    import urllib3
    http = urllib3.PoolManager()
    return http.request('GET', url, preload_content=False)

def with_requests(url):
    """Get a streaming response for the given event feed using requests."""
    import requests
    return requests.get(url, stream=True)

url = 'http://company.com/api/root_event_notification'
response = with_urllib3(url)  # or with_requests(url)
client = sseclient.SSEClient(response)
#print client.events()
for event in client.events():
    print "inside"
    pprint.pprint(json.loads(event.data))

run.sh

#!/bin/sh

while [ /usr/bin/true ]
do
    echo "Running sseevents.py"
    python sseevents.py  2>&1 | tee -a sseevents.log.txt

    echo "sleeping for 30 sec"
    sleep 30
done

SORTIE : -

Run call_notification.py on Terminal

node.js API OUTPUT

new_root_announced!
{'root_version': 'ABCD', 'BATS': '143'}

./run.sh --> DON'T SEE ABOVE EVENT below
Running sseevents.py
sleeping for 30 sec
Running sseevents.py
sleeping for 30 sec
Running sseevents.py
sleeping for 30 sec

Answer 1

Réponse très courte à votre question :

Le code du serveur ne renvoie pas de message SSE au client.

Pourquoi ? Parce que vous devez respecter le format ESS.

Selon JASON BUTZ dans Événements envoyés par le serveur avec Node

Vous devez envoyer un Connection : keep-alive pour s'assurer que le client garde la connexion ouverte. A Cache-Control doit être envoyé avec la valeur no-cache pour décourager la mise en cache des données. Enfin, le Content-Type doit être réglé sur text/event-stream .

Une fois tout cela fait, un nouveau trait ( \n ) doit être envoyé au client, puis les événements peuvent être envoyés. Les événements doivent être envoyés sous forme de chaînes, mais ce que contient cette chaîne n'a pas d'importance. Les chaînes JSON conviennent parfaitement.

Les données d'événement doivent être envoyées dans le format "data: <DATA TO SEND HERE>\n" .

Il est important de noter qu'à la fin de chaque ligne doit figurer un caractère de nouvelle ligne. Pour signifier la fin d'un événement, un caractère de fin de ligne supplémentaire doit également être ajouté.

Les lignes de données multiples conviennent parfaitement.

Longue réponse à votre question :

Selon Eric Bidelman dans html5rocks.com :

En communiquant à l'aide d'ESS, un serveur peut envoyer des données à votre application quand il le souhaite, sans avoir à faire une demande initiale. En d'autres termes, les mises à jour peuvent être diffusées du serveur au client au fur et à mesure qu'elles se produisent.

Mais, pour que cela se produise, le client doit "commencer" par le demander ET se préparer à recevoir une réponse. flux des messages (lorsqu'ils se produisent).

• Le "démarrage" se fait en appelant un point de terminaison de l'API SSE (dans votre cas, en appelant le code API Node.js).
• La préparation se fait en se préparant à traiter un flux de messages asynchrones.

Les ESS ouvrent un un seul canal unidirectionnel entre le serveur et le client.*

* L'accent est mis sur moi

Cela signifie que le serveur dispose d'un canal "direct" avec le client. Il n'est pas destiné à être "démarré" (ouvert) par un autre processus/code qui n'est pas le code "client".

D'après les commentaires de l'OP...

Comportement attendu (verbeux)

1. Un client Alice appelle le point de terminaison de l'API avec des paramètres {name: "Alice"} rien (de visible) ne se passe.

2. ...puis un client Bob appelle le point de terminaison de l'API avec des paramètres {name: "Bob"} le client Alice reçoit un SSE avec une charge utile {name: "Bob", says: "Hi"} .

3. ...puis un client Carol appelle le point de terminaison de l'API avec des paramètres {name: "Carol"} les clients Alice et Bob reçoivent chacun un SSE avec une charge utile {name: "Carol", says: "Hi"} .

4. ...et ainsi de suite. Chaque fois qu'un nouveau client appelle le point de terminaison de l'API avec des paramètres, tous les autres clients qui ont un canal "ouvert" recevront un SSE avec la nouvelle charge utile "Hi".

5. ...puis le client Bob se "déconnecte" du serveur, le client Alice, le client Carol et tous les clients qui ont un canal "ouvert" recevront un SSE avec une charge utile. {name: "Bob", says: "Bye"} .

6. ...et ainsi de suite. Chaque fois qu'un ancien client se "déconnecte" du serveur, tous les autres clients qui ont un canal "ouvert" reçoivent un SSE avec la nouvelle charge utile "Bye".

Comportement abstrait

• Chaque nouveau client qui demande à "ouvrir" un canal en envoyant des paramètres ou un ancien client qui se "déconnecte" du serveur, provoque un événement dans le serveur.
• Chaque fois qu'un tel événement se produit dans le serveur, celui-ci envoie un message SSE avec les paramètres et un message comme charge utile à tous les canaux "ouverts".

Note sur le blocage Chaque client avec un canal "ouvert" sera "bloqué" dans une boucle d'attente infinie pour que les événements se produisent. Il est de la responsabilité du client d'utiliser des techniques de code "threading" pour éviter les blocages.

Code

Votre client Python devrait "demander" de démarrer le canal unidirectionnel unique ET continuer à attendre jusqu'à ce que le canal soit fermé . Il ne doit pas se terminer et recommencer avec un autre canal. Il doit garder le même canal ouvert.

Du point de vue du réseau, ce sera comme une réponse "longue" qui ne se termine pas (jusqu'à ce que la messagerie ESS soit terminée). La réponse ne cesse d'arriver.

Votre code client Python s'en charge. J'ai noté que c'est le même exemple de code que celui utilisé dans le document Bibliothèque sseclient-py .

Code client pour Python 3.4

Pour inclure les paramètres que vous souhaitez envoyer au serveur, utilisez un peu de code de l'élément de menu Requests bibliothèque docs/#passing-paramètres-dans-urls .

Ainsi, en mélangeant ces échantillons, nous obtenons le code suivant comme client Python 3.4 :

import json
import pprint
import requests
import sseclient # sseclient-py

# change the name for each client
input_json = {'name':'Alice'}
#input_json = {'name':'Bob'}
#input_json = {'name':'Carol'}

url = 'http://company.com/api/root_event_notification'
stream_response = requests.get(url, params=input_json, stream=True)

client = sseclient.SSEClient(stream_response)

# Loop forever (while connection "open")
for event in client.events():
    print ("got a new event from server")
    pprint.pprint(event.data)

Code client pour Python 2.7

Pour inclure les paramètres que vous souhaitez envoyer au serveur, encodez-les dans l'URL en tant que paramètres de requête à l'aide de la commande urllib.urlencode() bibliothèque.

Faites la requête http avec urllib3.PoolManager().request() Vous obtiendrez ainsi une réponse en flux.

Notez que le sseclient renvoie les données de l'événement sous forme de chaîne unicode. Pour reconvertir l'objet JSON en objet python (avec des chaînes python) utilisez byteify une fonction personnalisée récursive ( merci à Mark Amery ).

Utilisez le code suivant comme client Python 2.7 :

import json
import pprint
import urllib
import urllib3
import sseclient # sseclient-py

# Function that returns byte strings instead of unicode strings
# Thanks to:
# [Mark Amery](https://stackoverflow.com/users/1709587/mark-amery)
def byteify(input):
    if isinstance(input, dict):
        return {byteify(key): byteify(value)
                for key, value in input.iteritems()}
    elif isinstance(input, list):
        return [byteify(element) for element in input]
    elif isinstance(input, unicode):
        return input.encode('utf-8')
    else:
        return input

# change the name for each client
input_json = {'name':'Alice'}
#input_json = {'name':'Bob'}
#input_json = {'name':'Carol'}

base_url = 'http://localhost:3000/api/root_event_notification'
url = base_url + '?' + urllib.urlencode(input_json)

http = urllib3.PoolManager()
stream_response = http.request('GET', url, preload_content=False)

client = sseclient.SSEClient(stream_response)

# Loop forever (while connection "open")
for event in client.events():
    print ("got a new event from server")
    pprint.pprint(byteify(json.loads(event.data)))

Maintenant, le code du serveur devrait :

1. émet un événement "hello" à l'intérieur du serveur, de sorte que autres clients écouter l'événement
2. "ouvrir" le canal
3. S'enregistrer pour écouter tous les événements possibles à l'intérieur du serveur (cela signifie garder le canal "ouvert" et ne rien envoyer entre les messages, juste garder le canal "ouvert").
   • Cela inclut l'émission d'un événement "d'au revoir" à l'intérieur du serveur. autres clients écouter l'événement QUAND le canal est fermé par le client/réseau (et enfin "wrap up").

Utilisez le code API Node.js suivant :

var EventEmitter = require('events').EventEmitter;
var myEmitter = new EventEmitter;

function registerEventHandlers(req, res) {
  // Save received parameters
  const myParams = req.query;

  // Define function that adds "Hi" and send a SSE formated message
  const sayHi = function(params) {
      params['says'] = "Hi";
      let payloadString = JSON.stringify(params);
      res.write(`data: ${payloadString}\n\n`);
  }

  // Define function that adds "Bye" and send a SSE formated message
  const sayBye = function(params) {
      params['says'] = "Bye";
      let payloadString = JSON.stringify(params);
      res.write(`data: ${payloadString}\n\n`);
  }

  // Register what to do when inside-server 'hello' event happens
  myEmitter.on('hello', sayHi);

  // Register what to do when inside-server 'goodbye' event happens
  myEmitter.on('goodbye', sayBye);

  // Register what to do when this channel closes
  req.on('close', () => {
      // Emit a server 'goodbye' event with "saved" params
      myEmitter.emit('goodbye', myParams);

      // Unregister this particular client listener functions
      myEmitter.off('hello', sayHi);
      myEmitter.off('goodbye', sayBye);
      console.log("<- close ", req.query);
  });
}

app.get("/api/root_event_notification", (req, res, next) => {
    console.log("open -> ", req.query);

    // Emit a inside-server 'hello' event with the received params
    myEmitter.emit('hello', req.query);

    // SSE Setup
    res.writeHead(200, {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
    });
    res.write('\n');

    // Register what to do when possible inside-server events happen
    registerEventHandlers(req, res);

    // Code execution ends here but channel stays open
    // Event handlers will use the open channel when inside-server events happen
})

...continuer à citer Eric Bidelman dans html5rocks.com :

L'envoi d'un flux d'événements depuis la source consiste à construire une réponse en texte clair, servie avec un Content-Type text/event-stream, qui suit le format SSE. Dans sa forme de base, la réponse doit contenir une ligne "data :", suivie de votre message, suivi de dos " \n "caractères pour terminer le flux

Dans le code client, le Bibliothèque sseclient-py se charge d'interpréter le format SSE de sorte qu'à chaque fois que l'option dos " \n Lorsque les caractères "itérables" arrivent, la bibliothèque "itère" un nouvel objet "itérable" (un nouvel événement) qui a la valeur "itérable". data avec le message envoyé par le serveur.

Voici comment j'ai testé le code

1. Serveur démarré avec le code API de Node.js
2. Exécutez un client avec seulement la ligne "Alice" non-commentée (Rien n'est encore vu sur la console de ce client).
3. Exécutez un deuxième client avec seulement la ligne "Bob" décommentée. La console du premier client "Alice" affiche : Bob disant "Salut" (Rien n'est encore visible sur la console client de Bob).
4. Exécutez un troisième client avec seulement la ligne "Carol" non-commentée. Les consoles d'Alice et de Bob s'affichent : Carol disant "Salut" (Rien n'est encore visible sur la console client de Carol).
5. Arrêter/supprimer le client de Bob. Les consoles d'Alice et de Carol s'affichent : Bob disant "Bye" .

Donc, le code fonctionne bien :)