receive method

@override Stream<AbstractMessage> receive()

Implementation

@override
Stream<AbstractMessage> receive() {
  final socket = _socket;
  if (socket == null) throw StateError('Transport is not open.');
  final openAttempt = _openAttempt;
  final onDisconnect = _onDisconnect!;
  final onConnectionLost = _onConnectionLost!;
  socket.done.then(
    (done) {
      if (!_goodbyeSent &&
          !_goodbyeReceived &&
          !onDisconnect.isCompleted &&
          !onConnectionLost.isCompleted) {
        onConnectionLost.complete();
      } else if (!onDisconnect.isCompleted) {
        onDisconnect.complete();
      }
    },
    onError: (error) {
      if (!_goodbyeSent &&
          !_goodbyeReceived &&
          !onDisconnect.isCompleted &&
          !onConnectionLost.isCompleted) {
        onConnectionLost.complete(error);
      }
    },
  );
  // TODO set keep alive to true
  //_socket.setOption(RawSocketOption.fromBool(??, SO_KEEPALIVE, true), true)
  return socket.expand((chunk) {
    if (openAttempt != _openAttempt || !identical(_socket, socket)) {
      return const <AbstractMessage>[];
    }
    return _consumeInboundChunk(chunk);
  });
}