Jak mogę kontrolować przepływ programu za pomocą zdarzeń i obietnic?

Question

Mam klasy tak:

import net from 'net'; 
import {EventEmitter} from 'events'; 
import Promise from 'bluebird'; 

class MyClass extends EventEmitter { 
    constructor(host = 'localhost', port = 10011) { 
     super(EventEmitter); 
     this.host = host; 
     this.port = port; 
     this.socket = null; 
     this.connect(); 
    } 
    connect() { 
     this.socket = net.connect(this.port, this.host); 
     this.socket.on('connect', this.handle.bind(this)); 
    } 
    handle(data) { 
     this.socket.on('data', data => { 

     }); 
    } 
    send(data) { 
     this.socket.write(data); 
    } 
} 

Jak bym włączyć metodę send do obietnicy, która zwraca wartość z gniazdka za data wydarzenia? Serwer wysyła dane tylko wtedy, gdy do nich są wysyłane dane, inne niż wiadomość o połączeniu, którą można łatwo zignorować.

Próbowałem coś takiego:

handle(data) { 
    this.socket.on('data', data => { 
     return this.socket.resolve(data); 
    }); 
    this.socket.on('error', this.socket.reject.bind(this)); 
} 
send(data) { 
    return new Promise((resolve, reject) => { 
     this.socket.resolve = resolve; 
     this.socket.reject = reject; 
     this.socket.write(data); 
    }); 
} 

Oczywiście to nie zadziała, ponieważ resolve/reject zastąpi siebie podczas łączenia i/lub dzwoniąc send wielokrotnie równolegle.

Istnieje również problem dwukrotnego równoległego wywoływania send i rozstrzyganie, która odpowiedź zostanie zwrócona jako pierwsza.

Mam obecnie implementację za pomocą kolejki i odradzam, ale czuje się niechlujnie, ponieważ kolejka jest stale sprawdzana.

Chciałbym być w stanie wykonać następujące czynności:

let c = new MyClass('localhost', 10011); 
c.send('foo').then(response => { 
    return c.send('bar', response.param); 
    //`response` should be the data returned from `this.socket.on('data')`. 
}).then(response => { 
    console.log(response); 
}).catch(error => console.log(error)); 

Wystarczy dodać, że nie mają żadnej kontroli nad danymi, które są odbierane, co oznacza, że ​​nie mogą być modyfikowane na zewnątrz strumień.

Edytuj: Wydaje się więc, że jest to całkiem niemożliwe, ponieważ TCP nie ma przepływu żądania-odpowiedzi. W jaki sposób można to zrealizować nadal przy użyciu obietnic, ale przy użyciu pojedynczego wykonania (jeden wniosek na raz) obietnicy łańcucha lub kolejki.

Answer 1

I destylowanej problem do minimum i uczynił z niego przeglądarkę runnable:

1. Klasa Socket jest wyśmiewany.
2. Usunięto informacje o porcie, hoście i dziedziczeniu z EventEmitter.

Rozwiązanie działa poprzez dołączanie nowych żądań do łańcucha obietnic, ale dopuszcza maksymalnie jedno otwarte lub nieodebrane żądanie w danym punkcie czasowym. .send zwraca nową obietnicę za każdym razem, gdy jest wywoływana, a klasa zajmuje się wszystkimi wewnętrznymi synchronizacjami. Tak więc .send może być wywoływana wiele razy, a poprawne uporządkowane (FIFO) przetwarzanie żądań jest gwarantowane. Dodatkową funkcją, którą dodałem, jest przycinanie łańcucha obietnic, jeśli nie ma oczekujących żądań.

---

Zastrzeżenie pominąłem obsługi błędów ogóle, butit powinna być dostosowana do konkretnego przypadku użycia tak.

---

DEMO

class SocketMock { 

    constructor(){ 
    this.connected = new Promise((resolve, reject) => setTimeout(resolve,200)); 
    this.listeners = { 
    // 'error' : [], 
    'data' : [] 
    } 
    } 

    send(data){ 

    console.log(`SENDING DATA: ${data}`); 
    var response = `SERVER RESPONSE TO: ${data}`; 
    setTimeout(() => this.listeners['data'].forEach(cb => cb(response)),    
       Math.random()*2000 + 250); 
    } 

    on(event, callback){ 
    this.listeners[event].push(callback); 
    } 

} 

class SingleRequestCoordinator { 

    constructor() { 
     this._openRequests = 0; 
     this.socket = new SocketMock(); 
     this._promiseChain = this.socket 
      .connected.then(() => console.log('SOCKET CONNECTED')); 
     this.socket.on('data', (data) => { 
     this._openRequests -= 1; 
     console.log(this._openRequests); 
     if(this._openRequests === 0){ 
      console.log('NO PENDING REQUEST --- trimming the chain'); 
      this._promiseChain = this.socket.connected 
     } 
     this._deferred.resolve(data); 
     }); 

    } 

    send(data) { 
     this._openRequests += 1; 
     this._promiseChain = this._promiseChain 
     .then(() => { 
      this._deferred = Promise.defer(); 
      this.socket.send(data); 
      return this._deferred.promise; 
     }); 
     return this._promiseChain; 
    } 
} 

var sender = new SingleRequestCoordinator(); 

sender.send('data-1').then(data => console.log(`GOT DATA FROM SERVER --- ${data}`)); 
sender.send('data-2').then(data => console.log(`GOT DATA FROM SERVER --- ${data}`)); 
sender.send('data-3').then(data => console.log(`GOT DATA FROM SERVER --- ${data}`)); 

setTimeout(() => sender.send('data-4') 
    .then(data => console.log(`GOT DATA FROM SERVER --- ${data}`)), 10000); 

Answer 2

Jeśli twoje połączenia send() są ze sobą pomieszane, należy zapisać je w pamięci podręcznej. Aby mieć pewność, że odebrana wiadomość pasuje do wysłanej, należy przypisać pewną unikatową wartość: id dla każdej wiadomości w polu danych.

Więc wiadomość nadawca będzie wyglądać następująco

class MyClass extends EventEmitter { 
    constructor() { 
    // [redacted] 
    this.messages = new Map(); 
    } 

    handle(data) { 
    this.socket.on('data', data => { 
     this.messages.get(data.id)(data); 
     this.messages.delete(data.id); 
    }); 
    } 

    send(data) { 
    return return new Promise((resolve, reject) => { 
     this.messages.set(data.id, resolve); 
     this.socket.write(data); 
    }); 
    } 
} 

Ten kod nie będzie sensowne kolejności komunikatów, a dostaniesz API, które chcesz.

Answer 3

socket.write(data[, encoding][, callback]) otrzymuje wywołanie zwrotne. Możesz odrzucić lub rozwiązać w tym oddzwonieniu.

class MyClass extends EventEmitter { 
    constructor(host = 'localhost', port = 10011) { 
    super(EventEmitter); 
    this.host = host; 
    this.port = port; 
    this.socket = null; 
    this.requests = null; 
    this.connect(); 
    } 
    connect() { 
    this.socket = net.connect(this.port, this.host); 
    this.socket.on('connect',() => { 
     this.requests = []; 
     this.socket.on('data', this.handle.bind(this)); 
     this.socket.on('error', this.error.bind(this)); 
    }); 
    } 
    handle(data) { 
    var [request, resolve, reject] = this.requests.pop(); 
    // I'm not sure what will happen with the destructuring if requests is empty 
    if(resolve) { 
     resolve(data); 
    } 
    } 
    error(error) { 
    var [request, resolve, reject] = this.requests.pop(); 
    if(reject) { 
     reject(error); 
    } 
    } 
    send(data) { 
    return new Promise((resolve, reject) => { 
     if(this.requests === null) { 
     return reject('Not connected'); 
     } 
     this.requests.push([data, resolve, reject]); 
     this.socket.write(data); 
    }); 
    } 
} 

Nie przetestowany i dlatego nie jest pewien podpisów metod, ale to jest podstawowy pomysł.

Zakłada się, że wystąpi jedno zdarzenie handle lub error na żądanie.

Im więcej o tym myślę, wydaje się to niemożliwe bez dodatkowych informacji w danych aplikacji, takich jak numery pakietów, które pasują do odpowiedzi na żądanie.

Sposób, w jaki jest teraz wdrażany (a także sposób, w jaki jest to w twoim pytaniu), nie jest nawet pewien, czy jedna odpowiedź będzie pasować dokładnie do jednego zdarzenia handle.