MessageListener for stateless transport.

This listener expect a handshake to precede each message.
11603
function StatelessListener(ptcl, readableFactory, opts) {
  MessageListener.call(this, ptcl, opts);
  var self = this;
  var readable;

  process.nextTick(function () {
    // Delay listening to allow handlers to be attached even if the factory is
    // purely synchronous.
    readable = readableFactory.call(this, function (err, writable) {
      if (err) {
        self.emit('error', err);
        // Since stateless listeners are only used once, it is safe to destroy.
        onFinish();
        return;
      }
      self._writable = writable.on('finish', onFinish);
      self.emit('_writable');
    }).on('data', onRequest)
      .on('end', onEnd);
  });

  function onRequest(obj) {
    var id = obj.id;
    var buf = Buffer.concat(obj.payload);
    var err = null;
    try {
      var parts = readHead(HANDSHAKE_REQUEST_TYPE, buf);
      var hreq = parts.head;
      var adapter = self._getAdapter(hreq);
    } catch (cause) {
      err = wrapError('invalid handshake request', cause);
    }

    if (err) {
      done(encodeError(err));
    } else {
      self._receive(parts.tail, adapter, done);
    }

    function done(resBuf) {
      if (!self._writable) {
        self.once('_writable', function () { done(resBuf); });
        return;
      }
      var hres = self._createHandshakeResponse(err, hreq);
      self.emit('handshake', hreq, hres);
      var payload = [
        HANDSHAKE_RESPONSE_TYPE.toBuffer(hres),
        resBuf
      ];
      self._writable.write({id: id, payload: payload});
      if (self._endWritable) {
        self._writable.end();
      }
    }
  }

  function onEnd() { self.destroy(); }

  function onFinish() {
    if (readable) {
      readable
        .removeListener('data', onRequest)
        .removeListener('end', onEnd);
    }
    self.destroy(true);
  }
}
788899
public synchronized void addProtocol(String productString, HttpUpgradeListener openListener, final HttpUpgradeHandshake handshake) {
        addProtocol(productString, openListener, null, handshake);
    }
1036789
public void addListeners()
    {
        super.addListeners();
        this.getMainRecord().addListener(new MessageDetailDefTransHandler(null));
        String strManualTransportID = Integer.toString(((ReferenceField)this.getMainRecord().getField(MessageDetail.MESSAGE_TRANSPORT_ID)).getIDFromCode(MessageTransport.MANUAL));
        this.getMainRecord().getField(MessageDetail.MESSAGE_TRANSPORT_ID).setDefault(strManualTransportID);
    }
932188
@Override
    public void sendSilenceMessage(
                                   long startStamp,
                                   long endStamp,
                                   long completedPrefix,
                                   boolean requestedOnly,
                                   int priority,
                                   Reliability reliability,
                                   SIBUuid12 stream)
    {
        // NOP, the PubSubInputHandler doesn't originate
        // silence.  The associated output handlers do it
        // for us.
    }
611170
public void handshake(NextFilter nextFilter) throws SSLException {
        for (;;) {
            switch (handshakeStatus) {
                case FINISHED :
                    session.setAttribute(
                            SslFilter.SSL_SESSION, sslEngine.getSession());
                    handshakeComplete = true;

                    if (logger.isDebugEnabled()) {
                        SSLSession sslSession = sslEngine.getSession();
                        logger.debug(String.format("SSL session ID %s on transport session #%d %s: cipher %s, app buffer size %d, packet buffer size %d", 
                            sslSession.getId(), session.getId(), session, sslSession.getCipherSuite(), sslSession.getApplicationBufferSize(), sslSession.getPacketBufferSize()));
                    }

                    if (!initialHandshakeComplete
                            && session.containsAttribute(SslFilter.USE_NOTIFICATION)) {
                        // SESSION_SECURED is fired only when it's the first handshake.
                        // (i.e. renegotiation shouldn't trigger SESSION_SECURED.)
                        initialHandshakeComplete = true;
                        scheduleMessageReceived(nextFilter,
                                SslFilter.SESSION_SECURED);
                    }
                    
                    return;
                    
                case NEED_TASK :
                    handshakeStatus = doTasks();
                    break;
                    
                case NEED_UNWRAP :
                    // we need more data read
                    SSLEngineResult.Status status = unwrapHandshake(nextFilter);
                    
                    if (status == SSLEngineResult.Status.BUFFER_UNDERFLOW &&
                            handshakeStatus != SSLEngineResult.HandshakeStatus.FINISHED ||
                            isInboundDone()) {
                        // We need more data or the session is closed
                        return;
                    }
                    
                    break;

                case NEED_WRAP :
                    // First make sure that the out buffer is completely empty. Since we
                    // cannot call wrap with data left on the buffer
                    if (outNetBuffer != null && outNetBuffer.hasRemaining()) {
                        return;
                    }

                    SSLEngineResult result;
                    createOutNetBuffer(0);
                    
                    for (;;) {
                        result = sslEngine.wrap(emptyBuffer.buf(), outNetBuffer.buf());
                        if (result.getStatus() == SSLEngineResult.Status.BUFFER_OVERFLOW) {
                            outNetBuffer.capacity(outNetBuffer.capacity() << 1, allocator);
                            outNetBuffer.limit(outNetBuffer.capacity());
                        } else {
                            break;
                        }
                    }

                    outNetBuffer.flip();
                    handshakeStatus = result.getHandshakeStatus();
                    writeNetBuffer(nextFilter);
                    break;
            
                default :
                    throw new IllegalStateException("Invalid Handshaking State"
                            + handshakeStatus);
            }
        }
    }
769170
void sendNextProtocol(NextProtoNego.Provider provider) throws IOException
    {
        if (isInitialHandshake && provider != null)
        {
            if (protocols == null)
            {
                if (NextProtoNego.debug)
                    System.err.println(new StringBuilder("[C] NPN not received for ").append(conn != null ? conn : engine));
                ((NextProtoNego.ClientProvider)provider).unsupported();
            }
            else
            {
                try
                {
                    String protocol = ((NextProtoNego.ClientProvider)provider).selectProtocol(protocols);
                    if (NextProtoNego.debug)
                        System.err.println(new StringBuilder("[C] NPN selected '").append(protocol).append("' for ").append(conn != null ? conn : engine));
                    new NextProtocolMessage(protocol).write(output);
                    output.flush();
                }
                catch(Throwable t)
                {
                    fatalSE(Alerts.alert_handshake_failure, "NPN protocol selection failure", t);
                }
            }
        }
    }
97753
def MessageReceived(self, m):
        
        if m.Command == 'verack':
            # only respond with a verack when we connect to another client, not when a client connected to us or
            # we might end up in a verack loop
            if self.incoming_client:
                if self.expect_verack_next:
                    self.expect_verack_next = False
            else:
                self.HandleVerack()
        elif m.Command == 'version':
            self.HandleVersion(m.Payload)
        elif m.Command == 'getaddr':
            self.SendPeerInfo()
        elif m.Command == 'getdata':
            self.HandleGetDataMessageReceived(m.Payload)
        elif m.Command == 'getblocks':
            self.HandleGetBlocksMessageReceived(m.Payload)
        elif m.Command == 'inv':
            self.HandleInvMessage(m.Payload)
        elif m.Command == 'block':
            self.HandleBlockReceived(m.Payload)
        elif m.Command == 'getheaders':
            self.HandleGetHeadersMessageReceived(m.Payload)
        elif m.Command == 'headers':
            self.HandleBlockHeadersReceived(m.Payload)
        elif m.Command == 'addr':
            self.HandlePeerInfoReceived(m.Payload)
        else:
            logger.debug(f"{self.prefix} Command not implemented: {m.Command}")
625760
public synchronized void addProtocol(String productString, ChannelListener<? super StreamConnection> openListener, final HttpUpgradeHandshake handshake) {
        addProtocol(productString, null, openListener, handshake);
    }
1302555
function _handshake(handshakeProps, handshakeCallback) {
        if (_isFunction(handshakeProps)) {
            handshakeCallback = handshakeProps;
            handshakeProps = undefined;
        }

        _clientId = null;

        _clearSubscriptions();

        // Reset the transports if we're not retrying the handshake
        if (_isDisconnected()) {
            _transports.reset(true);
            _updateAdvice(_config.advice);
        } else {
            // We are retrying the handshake, either because another handshake failed
            // and we're backing off, or because the server timed us out and asks us to
            // re-handshake: in both cases, make sure that if the handshake succeeds
            // the next action is a connect.
            _updateAdvice(_cometd._mixin(false, _advice, {reconnect: 'retry'}));
        }

        _batch = 0;

        // Mark the start of an internal batch.
        // This is needed because handshake and connect are async.
        // It may happen that the application calls init() then subscribe()
        // and the subscribe message is sent before the connect message, if
        // the subscribe message is not held until the connect message is sent.
        // So here we start a batch to hold temporarily any message until
        // the connection is fully established.
        _internalBatch = true;

        // Save the properties provided by the user, so that
        // we can reuse them during automatic re-handshake
        _handshakeProps = handshakeProps;
        _handshakeCallback = handshakeCallback;

        var version = '1.0';

        // Figure out the transports to send to the server
        var url = _cometd.getURL();
        var transportTypes = _transports.findTransportTypes(version, _crossDomain, url);

        var bayeuxMessage = {
            id: _nextMessageId(),
            version: version,
            minimumVersion: version,
            channel: '/meta/handshake',
            supportedConnectionTypes: transportTypes,
            advice: {
                timeout: _advice.timeout,
                interval: _advice.interval
            }
        };
        // Do not allow the user to override important fields.
        var message = _cometd._mixin(false, {}, _handshakeProps, bayeuxMessage);

        // Save the callback.
        _cometd._putCallback(message.id, handshakeCallback);

        // Pick up the first available transport as initial transport
        // since we don't know if the server supports it
        if (!_transport) {
            _transport = _transports.negotiateTransport(transportTypes, version, _crossDomain, url);
            if (!_transport) {
                var failure = 'Could not find initial transport among: ' + _transports.getTransportTypes();
                _cometd._warn(failure);
                throw failure;
            }
        }

        _cometd._debug('Initial transport is', _transport.getType());

        // We started a batch to hold the application messages,
        // so here we must bypass it and send immediately.
        _setStatus('handshaking');
        _cometd._debug('Handshake sent', message);
        _send(false, [message], false, 'handshake');
    }
1275483
def begin(self):
        
        log.debug("handshake (version %s)" % PROTOCOL_VERSION)
        version = Version()
        version.services = 0    # can't send blocks
        log.debug("send Version")
        self.send_message(version)
1197370
private boolean handshake()
    {
        assert (handshaking);
        assert (greetingRecv.position() < greetingSize);

        final Mechanisms mechanism = options.mechanism;
        assert (mechanism != null);

        //  Position of the version field in the greeting.
        final int revisionPos = SIGNATURE_SIZE;

        //  Make sure batch sizes match large buffer sizes
        final int inBatchSize = Math.max(options.rcvbuf, Config.IN_BATCH_SIZE.getValue());
        final int outBatchSize = Math.max(options.sndbuf, Config.OUT_BATCH_SIZE.getValue());

        //  Receive the greeting.
        while (greetingRecv.position() < greetingSize) {
            final int n = read(greetingRecv);
            if (n == 0) {
                error(ErrorReason.CONNECTION);
                return false;
            }
            if (n == -1) {
                if (!errno.is(ZError.EAGAIN)) {
                    error(ErrorReason.CONNECTION);
                }
                return false;
            }

            //  We have received at least one byte from the peer.
            //  If the first byte is not 0xff, we know that the
            //  peer is using unversioned protocol.
            if ((greetingRecv.get(0) & 0xff) != 0xff) {
                // If this first byte is not %FF,
                // then the other peer is using ZMTP 1.0.
                break;
            }

            if (greetingRecv.position() < SIGNATURE_SIZE) {
                continue;
            }

            //  Inspect the right-most bit of the 10th byte (which coincides
            //  with the 'flags' field if a regular message was sent).
            //  Zero indicates this is a header of identity message
            //  (i.e. the peer is using the unversioned protocol).
            if ((greetingRecv.get(9) & 0x01) != 0x01) {
                break;
            }

            // If the least significant bit is 1, the peer is using ZMTP 2.0 or later
            // and has sent us the ZMTP signature.

            int outpos = greetingSend.position();
            //  The peer is using versioned protocol.
            //  Send the major version number.
            if (greetingSend.limit() == SIGNATURE_SIZE) {
                if (outsize == 0) {
                    ioObject.setPollOut(handle);
                }
                greetingSend.limit(SIGNATURE_SIZE + 1);
                greetingSend.put(revisionPos, Protocol.V3.revision); //  Major version number
                outsize += 1;
            }

            if (greetingRecv.position() > SIGNATURE_SIZE) {
                if (greetingSend.limit() == SIGNATURE_SIZE + 1) {
                    if (outsize == 0) {
                        ioObject.setPollOut(handle);
                    }
                    // We read a further byte, which indicates the ZMTP version.
                    byte protocol = greetingRecv.get(revisionPos);
                    if (protocol == Protocol.V1.revision || protocol == Protocol.V2.revision) {
                        // If this is V1 or V2, we have a ZMTP 2.0 peer.
                        greetingSend.limit(V2_GREETING_SIZE);
                        greetingSend.position(SIGNATURE_SIZE + 1);
                        greetingSend.put((byte) options.type); // Socket type
                        outsize += 1;
                    }
                    else {
                        // If this is 3 or greater, we have a ZMTP 3.0 peer.
                        greetingSend.limit(V3_GREETING_SIZE);
                        greetingSend.position(SIGNATURE_SIZE + 1);
                        greetingSend.put((byte) 0); //  Minor version number
                        outsize += 1;
                        greetingSend.mark();
                        greetingSend.put(new byte[20]);

                        assert (mechanism == Mechanisms.NULL || mechanism == Mechanisms.PLAIN
                                || mechanism == Mechanisms.CURVE || mechanism == Mechanisms.GSSAPI);
                        greetingSend.reset();
                        greetingSend.put(mechanism.name().getBytes(ZMQ.CHARSET));
                        greetingSend.reset();
                        greetingSend.position(greetingSend.position() + 20);
                        outsize += 20;
                        greetingSend.put(new byte[32]);
                        outsize += 32;

                        greetingSize = V3_GREETING_SIZE;
                    }
                }
            }
            greetingSend.position(outpos);
        }

        //  Is the peer using the unversioned protocol?
        //  If so, we send and receive rest of identity
        //  messages.
        if ((greetingRecv.get(0) & 0xff) != 0xff || (greetingRecv.get(9) & 0x01) == 0) {
            // If this first byte is %FF, then we read nine further bytes,
            // and inspect the last byte (the 10th in total).
            // If the least significant bit is 0, then the other peer is using ZMTP 1.0.
            if (session.zapEnabled()) {
                // reject ZMTP 1.0 connections if ZAP is enabled
                error(ErrorReason.PROTOCOL);
                return false;
            }

            zmtpVersion = Protocol.V0;

            encoder = new V1Encoder(errno, outBatchSize);
            decoder = new V1Decoder(errno, inBatchSize, options.maxMsgSize, options.allocator);

            //  We have already sent the message header.
            //  Since there is no way to tell the encoder to
            //  skip the message header, we simply throw that
            //  header data away.
            final int headerSize = options.identitySize + 1 >= 255 ? 10 : 2;
            ByteBuffer tmp = ByteBuffer.allocate(headerSize);

            //  Prepare the identity message and load it into encoder.
            //  Then consume bytes we have already sent to the peer.
            Msg txMsg = new Msg(options.identitySize);
            txMsg.put(options.identity, 0, options.identitySize);
            encoder.loadMsg(txMsg);

            ValueReference<ByteBuffer> bufferp = new ValueReference<>(tmp);
            int bufferSize = encoder.encode(bufferp, headerSize);
            assert (bufferSize == headerSize);

            //  Make sure the decoder sees the data we have already received.
            decodeDataAfterHandshake(0);

            //  To allow for interoperability with peers that do not forward
            //  their subscriptions, we inject a phantom subscription message
            //  message into the incoming message stream.
            if (options.type == ZMQ.ZMQ_PUB || options.type == ZMQ.ZMQ_XPUB) {
                subscriptionRequired = true;
            }

            //  We are sending our identity now and the next message
            //  will come from the socket.
            nextMsg = pullMsgFromSession;
            //  We are expecting identity message.
            processMsg = processIdentity;

        }
        else if (greetingRecv.get(revisionPos) == Protocol.V1.revision) {
            //  ZMTP/1.0 framing.

            zmtpVersion = Protocol.V1;

            if (session.zapEnabled()) {
                // reject ZMTP 1.0 connections if ZAP is enabled
                error(ErrorReason.PROTOCOL);
                return false;
            }
            encoder = new V1Encoder(errno, outBatchSize);
            decoder = new V1Decoder(errno, inBatchSize, options.maxMsgSize, options.allocator);

            decodeDataAfterHandshake(V2_GREETING_SIZE);
        }
        else if (greetingRecv.get(revisionPos) == Protocol.V2.revision) {
            //  ZMTP/2.0 framing.

            zmtpVersion = Protocol.V2;

            if (session.zapEnabled()) {
                // reject ZMTP 2.0 connections if ZAP is enabled
                error(ErrorReason.PROTOCOL);
                return false;
            }
            encoder = new V2Encoder(errno, outBatchSize);
            decoder = new V2Decoder(errno, inBatchSize, options.maxMsgSize, options.allocator);

            decodeDataAfterHandshake(V2_GREETING_SIZE);
        }
        else {
            zmtpVersion = Protocol.V3;

            encoder = new V2Encoder(errno, outBatchSize);
            decoder = new V2Decoder(errno, inBatchSize, options.maxMsgSize, options.allocator);

            greetingRecv.position(V2_GREETING_SIZE);
            if (mechanism.isMechanism(greetingRecv)) {
                this.mechanism = mechanism.create(session, peerAddress, options);
            }
            else {
                error(ErrorReason.PROTOCOL);
                return false;
            }

            nextMsg = nextHandshakeCommand;
            processMsg = processHandshakeCommand;
        }

        // Start polling for output if necessary.
        if (outsize == 0) {
            ioObject.setPollOut(handle);
        }

        //  Handshaking was successful.
        //  Switch into the normal message flow.
        handshaking = false;

        if (hasHandshakeTimer) {
            ioObject.cancelTimer(HANDSHAKE_TIMER_ID);
            hasHandshakeTimer = false;
        }

        socket.eventHandshaken(endpoint, zmtpVersion.ordinal());

        return true;
    }
377704
@Override
    public void doMessageReceivedListeners(final long sessionId, final long sessionReadBytes, final Object message) {
        runManagementTask(new Runnable() {
            @Override
            public void run() {
                try {
                    // The particular management listeners change on strategy, so get them here.
                    for (final GatewayManagementListener listener : getManagementListeners()) {
                        listener.doMessageReceived(GatewayManagementBeanImpl.this, sessionId);
                    }

                    markChanged();  // mark ourselves as changed, possibly tell listeners
                } catch (Exception ex) {
                    logger.warn("Error during messageReceived gateway listener notifications:", ex);
                }
            }
        });
    }
368801
void notifyListenersOnStateChange() {
        LOGGER.debug("Notifying connection listeners about state change to {}", state);
        
        for (ConnectionListener listener : connectionListeners) {
            switch (state) {
                case CONNECTED:
                    listener.onConnectionEstablished(connection);
                    break;
                case CONNECTING:
                    listener.onConnectionLost(connection);
                    break;
                case CLOSED:
                    listener.onConnectionClosed(connection);
                    break;
                default:
                    break;
            }
        }
    }
103398
private void notifyListenersBeforeWrite(ModbusMessage msg) {
        synchronized (listeners) {
            for (AbstractSerialTransportListener listener : listeners) {
                listener.beforeMessageWrite(commPort, msg);
            }
        }
    }
1259886
void notifyListenersNetworkUneachable() {
        // NOTE: synchronized(listener) causes deadlock with listeners and Replicator.lock.
        final Set<NetworkReachabilityListener> copy = new HashSet<>(listeners);
        for (NetworkReachabilityListener listener : copy) {
            if (listener != null) { listener.networkUnreachable(); }
        }
    }
272053
public void addListener(IsSimplePromiseResponseHandler<T> listener) {
    if (handlers == null) {
      handlers = new LinkedList<>();
    }

    handlers.add(listener);

    if (response != null || exception != null) {
      listener.onResponse(this);
    }
  }
234246
@SuppressWarnings("unchecked")
    @Override
    public void start(final Listener<?> listener, final Infrastructure infra) throws MessageTransportException {
        this.listener = (Listener<Object>) listener;
    }