Comment utiliser RxJs avec la Prise.IO sur l'événement

Je veux utiliser RxJS à l'intérieur de mon socket.on('sense',function(data){});. Je suis coincé et se confond avec très peu de documentation disponible et de mon manque de compréhension RxJS. Voici mon problème.

J'ai un distSensor.js qui a une fonction pingEnd()

function pingEnd(x){
socket.emit("sense", dist); //pingEnd is fired when an Interrupt is generated.
}

À l'intérieur de mon App.js j'ai

io.on('connection', function (socket) {
    socket.on('sense', function (data) {
        //console.log('sense from App4 was called ' + data);
    });
});

Le sens de la fonction de beaucoup de données de capteur qui je veux filtrer à l'aide de RxJS et je ne sais pas que dois-je faire pour utiliser RxJs ici. Tous les pointeurs de droite docs ou de l'échantillon de l'aide.

InformationsquelleAutor Mitul | 2015-04-23

16

Je pense que vous pouvez utiliser Rx.Observable.fromEvent (https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/fromevent.md).

Voici comment j'ai fait quelque chose de similaire à l'aide de Bacon.js qui a une très similaires API: https://github.com/raimohanska/bacon-minsk-2015/blob/gh-pages/server.js#L13

Donc dans Bacon.js elle irait comme
```
io.on('connection', function(socket){
  Bacon.fromEvent(socket, "sense")
    .filter(function(data) { return true })
    .forEach(function(data) { dealWith(data) })
})
```
Et dans RxJs vous auriez remplacer Bacon.fromEvent avec Rx.Observable.fromEvent.
- Veuillez supprimer le "je pense", car il fonctionne très bien. Également remplacer les Lardons par Rx.Observable comme vous l'avez déjà décrite.
InformationsquelleAutor raimohanska
16

J'ai vécu de drôles de questions à l'aide de la fromEvent méthode, donc je préfère juste de créer mon propre Observables:
```
function RxfromIO (io, eventName) {
  return Rx.Observable.create(observer => {
    io.on(eventName, (data) => {
        observer.onNext(data)
    });
    return {
        dispose : io.close
    }
});
```
Je peux alors l'utiliser comme ceci:
```
let $connection = RxfromIO(io, 'connection');
```
- Utilisez simplement fromEvent pas besoin d'écrire ce code supplémentaire! const connexion$ = Rx.Observables.fromEvent(io, 'connexion');
InformationsquelleAutor Morten Poulsen
9

Vous pouvez créer un Observables comme suit:
```
var senses = Rx.Observable.fromEventPattern(
    function add (h) {
      socket.on('sense',h);
    }
  );
```
Ensuite utiliser senses comme tous les autres Observables.
- Quelle est la fonction de l'ajouter(h) méthode? C'est que la fonction de la partie de RxJs?
- Je suis juste de nommage de la fonction, il suffit de le faire: function(h){...}.
- fromEvent n'a pas fonctionné pour moi. Et j'ai fini avec cette solution au lieu de cela. En TS ses: let obs = Observable.fromEventPattern(h => this.socket.on('my_event', h));
- Utilisez simplement fromEvent pas besoin d'écrire ce code supplémentaire! const sense$ = Rx.Observable.fromEvent(socket, 'sense');
InformationsquelleAutor Xesued

Simplement utiliser fromEvent(). Voici un exemple complet dans Node.js mais fonctionne de la même façon dans le navigateur. Notez que j'utilise first() et takeUntil() pour éviter une fuite de mémoire: first() écoute seulement d'un événement, puis se termine. Maintenant utiliser takeUntil() sur tous les autres socket-les événements de vous écouter afin de les observables complète à la déconnexion:

const app = require('express')();
const server = require('http').createServer(app);
const io = require('socket.io')(server);
const Rx = require('rxjs/Rx');

connection$ = Rx.Observable.fromEvent(io, 'connection');

connection$.subscribe(socket => {
    console.log(`Client connected`);

    //Observables
    const disconnect$ = Rx.Observable.fromEvent(socket, 'disconnect').first();
    const message$ = Rx.Observable.fromEvent(socket, 'message').takeUntil(disconnect$);

    //Subscriptions
    message$.subscribe(data => {
        console.log(`Got message from client with data: ${data}`);
        io.emit('message', data); //Emit to all clients
    });

    disconnect$.subscribe(() => {
        console.log(`Client disconnected`);
    })
});

server.listen(3000);

Bonjour @Mick, un excellent exemple, mais que diriez-côté client? Je suis aux prises avec la façon de déconnexion lors de la désinscription est appelé au cours de la fromEvent observables.

InformationsquelleAutor Mick

ES6 un liner que j'utilise, à l'aide de ES7 lier syntaxe:

(lire $ comme stream)

import { Observable } from 'rxjs'

//create socket

const message$ = Observable.create($ => socket.on('message', ::$.next))
//translates to: Observable.create($ => socket.on('message', $.next.bind(this)))

//filter example
const subscription = message$
  .filter(message => message.text !== 'spam')
  //or .filter(({ text }) => text !== 'spam')
  .subscribe(::console.log)

InformationsquelleAutor woudsma

Vous pouvez utiliser rxjs-dom,

Rx.DOM.fromWebSocket(url, protocol, [openObserver], [closeObserver])


//an observer for when the socket is open
var openObserver = Rx.Observer.create(function(e) {
  console.info('socket open');

  //Now it is safe to send a message
  socket.onNext('test');
});

//an observer for when the socket is about to close
var closingObserver = Rx.Observer.create(function() {
  console.log('socket is about to close');
});

//create a web socket subject
socket = Rx.DOM.fromWebSocket(
  'ws://echo.websocket.org',
  null, //no protocol
  openObserver,
  closingObserver);

//subscribing creates the underlying socket and will emit a stream of incoming
//message events
socket.subscribe(
  function(e) {
    console.log('message: %s', e.data);
  },
  function(e) {
    //errors and "unclean" closes land here
    console.error('error: %s', e);
  },
  function() {
    //the socket has been closed
    console.info('socket closed');
  }
);

Il semble que les Rx.DOM.fromWebSocket ne pouvez pas l'utiliser avec la prise.io-serveur. Rx.DOM.fromWebSocket utilisation standard du w3c API websocket.

InformationsquelleAutor serkan

Vous devez vous connecter pour publier un commentaire.