Process the incoming message, calling the session decoder. As the incoming
buffer might contains more than one messages, we have to loop until the decoder
throws an exception.

while ( buffer not empty )
try
decode ( buffer )
catch
break;
773722
@Override
    public void messageReceived(NextFilter nextFilter, IoSession session,
            Object message) throws Exception {
        LOGGER.debug("Processing a MESSAGE_RECEIVED for session {}", session.getId());

        if (!(message instanceof IoBuffer)) {
            nextFilter.messageReceived(session, message);
            return;
        }

        IoBuffer in = (IoBuffer) message;
        ProtocolDecoder decoder = getDecoder(session);
        ProtocolDecoderOutput decoderOut = getDecoderOut(session, nextFilter);
        IoSessionEx sessionEx = (IoSessionEx)session;

        Thread ioThread = sessionEx.getIoThread();

        // Loop until we don't have anymore byte in the buffer,
        // or until the decoder throws an unrecoverable exception or
        // can't decoder a message, because there are not enough
        // data in the buffer
        // Note: break out to let old I/O thread unwind after deregistering
        while (in.hasRemaining()) {
            // If the session is realigned, let the new thread deal with the decoding
            if (sessionEx.getIoThread() != ioThread) {
                if (LOGGER.isTraceEnabled()) {
                    LOGGER.trace(format("Decoding for session=%s will be continued by new thread=%s old thread=%s",
                            session, sessionEx.getIoThread(), ioThread));
                }
                break;
            }
            int oldPos = in.position();
            try {
                synchronized (decoderOut) {
                    // Call the decoder with the read bytes
                    decoder.decode(session, in, decoderOut);

                    // Finish decoding if no exception was thrown.
                    decoderOut.flush(nextFilter, session);
                }
            } catch (Throwable t) {
                ProtocolDecoderException pde;
                if (t instanceof ProtocolDecoderException) {
                    pde = (ProtocolDecoderException) t;
                } else {
                    pde = new ProtocolDecoderException(t);
                }

                if (pde.getHexdump() == null) {
                    // Generate a message hex dump
                    int curPos = in.position();
                    in.position(oldPos);
                    pde.setHexdump(in.getHexDump());
                    in.position(curPos);
                }

                // Fire the exceptionCaught event.
                synchronized (decoderOut) {
                    decoderOut.flush(nextFilter, session);
                }
                nextFilter.exceptionCaught(session, pde);

                // Retry only if the type of the caught exception is
                // recoverable and the buffer position has changed.
                // We check buffer position additionally to prevent an
                // infinite loop.
                if (!(t instanceof RecoverableProtocolDecoderException) ||
                        (in.position() == oldPos)) {
                    break;
                }
            }
        }
    }
1156797
boolean processNextFrame() throws IOException {
    try {
      reader.processNextFrame();
      return receivedCloseCode == -1;
    } catch (Exception e) {
      failWebSocket(e, null);
      return false;
    }
  }
507053
def handle(self):
        
        logger = StreamHandler.logger
        logger.debug("handling requests with message handler %s " % StreamHandler.message_handler.__class__.__name__)

        message_handler = StreamHandler.message_handler

        try:
            while True:
                logger.debug('waiting for more data')
                if not message_handler.handle(self.request, StreamHandler.BUFFER_SIZE):
                    break

            logger.warning("connection closed from %s" % (self.client_address[0]))
            self.request.close()
        except:
            logger.exception("connection closed from %s" % (self.client_address[0]))
        finally:
            self.request.close()
1213684
protected void processReceivedMessage(String channel, byte[] message) {

		// Parse message
		Tree data;
		try {
			data = serializer.read(message);
		} catch (Exception cause) {
			logger.warn("Unable to parse incoming message!", cause);
			return;
		}

		// Debug
		if (debug && (debugHeartbeats || !channel.endsWith(heartbeatChannel))) {
			logger.info("Message received from channel \"" + channel + "\":\r\n" + data);
		}

		// Send message to proper component
		try {

			// Get "sender" property
			String sender = data.get("sender", "");
			if (sender == null || sender.isEmpty()) {
				logger.warn("Missing \"sender\" property:\r\n" + data);
				return;
			}
			if (sender.equals(nodeID)) {

				// It's our message
				return;
			}

			// Incoming response
			if (channel.equals(responseChannel)) {
				registry.receiveResponse(data);
				return;
			}

			// Incoming event
			if (channel.equals(eventChannel)) {
				eventbus.receiveEvent(data);
				return;
			}

			// Incoming request
			if (channel.equals(requestChannel)) {
				registry.receiveRequest(data);
				return;
			}

			// HeartBeat packet
			if (channel.endsWith(heartbeatChannel)) {

				// Get node container
				NodeDescriptor node = nodes.get(sender);
				if (node == null) {

					// Unknown node -> send discover packet
					sendDiscoverPacket(channel(PACKET_DISCOVER, sender));
					return;
				}
				int cpu = data.get("cpu", 0);

				// Update CPU info
				boolean offline;
				node.writeLock.lock();
				try {
					node.updateCpu(cpu);
					offline = node.offlineSince > 0;
				} finally {
					node.writeLock.unlock();
				}
				if (offline) {

					// Offline node -> send discover packet
					sendDiscoverPacket(channel(PACKET_DISCOVER, sender));
				}
				return;
			}

			// Info packet
			if (channel.equals(infoChannel) || channel.equals(infoBroadcastChannel)) {

				// Register services and listeners
				data.put("seq", System.currentTimeMillis());
				data.put("port", 1);
				updateNodeInfo(sender, data);
				return;
			}

			// Discover packet
			if (channel.equals(discoverChannel) || channel.equals(discoverBroadcastChannel)) {

				// Send node desriptor to the sender
				sendInfoPacket(channel(PACKET_INFO, sender));
				return;
			}

			// Ping packet
			if (channel.equals(pingChannel)) {
				sendPongPacket(channel(PACKET_PONG, sender), data);
				return;
			}

			// Pong packet
			if (channel.equals(pongChannel)) {
				registry.receivePong(data);
				return;
			}

			// Disconnect packet
			if (channel.equals(disconnectChannel)) {

				// Switch to offline
				NodeDescriptor node = nodes.get(sender);
				if (node == null) {
					return;
				}
				boolean disconnected = false;
				node.writeLock.lock();
				try {
					if (node.markAsOffline()) {

						// Remove remote actions and listeners
						registry.removeActions(sender);
						eventbus.removeListeners(sender);
						disconnected = true;

					}
				} finally {
					node.writeLock.unlock();
				}
				if (node != null && disconnected) {

					// Notify listeners (not unexpected disconnection)
					logger.info("Node \"" + sender + "\" disconnected.");
					broadcastNodeDisconnected(node.info, false);
				}
				return;
			}

		} catch (Exception cause) {
			logger.warn("Unable to process incoming message!", cause);
		}
	}
14372
public void decode(ByteBuffer buffer, boolean moreData) throws HpackException {
        while (buffer.hasRemaining()) {
            int originalPos = buffer.position();
            byte b = buffer.get();
            if ((b & 0b10000000) != 0) {
                first = false;
                //if the first bit is set it is an indexed header field
                buffer.position(buffer.position() - 1); //unget the byte
                int index = Hpack.decodeInteger(buffer, 7); //prefix is 7
                if (index == -1) {
                    if(!moreData) {
                        throw UndertowMessages.MESSAGES.hpackFailed();
                    }
                    buffer.position(originalPos);
                    return;
                } else if(index == 0) {
                    throw UndertowMessages.MESSAGES.zeroNotValidHeaderTableIndex();
                }
                handleIndex(index);
            } else if ((b & 0b01000000) != 0) {
                first = false;
                //Literal Header Field with Incremental Indexing
                HttpString headerName = readHeaderName(buffer, 6);
                if (headerName == null) {
                    if(!moreData) {
                        throw UndertowMessages.MESSAGES.hpackFailed();
                    }
                    buffer.position(originalPos);
                    return;
                }
                String headerValue = readHpackString(buffer);
                if (headerValue == null) {
                    if(!moreData) {
                        throw UndertowMessages.MESSAGES.hpackFailed();
                    }
                    buffer.position(originalPos);
                    return;
                }
                headerEmitter.emitHeader(headerName, headerValue, false);
                addEntryToHeaderTable(new HeaderField(headerName, headerValue));
            } else if ((b & 0b11110000) == 0) {
                first = false;
                //Literal Header Field without Indexing
                HttpString headerName = readHeaderName(buffer, 4);
                if (headerName == null) {
                    if(!moreData) {
                        throw UndertowMessages.MESSAGES.hpackFailed();
                    }
                    buffer.position(originalPos);
                    return;
                }
                String headerValue = readHpackString(buffer);
                if (headerValue == null) {
                    if(!moreData) {
                        throw UndertowMessages.MESSAGES.hpackFailed();
                    }
                    buffer.position(originalPos);
                    return;
                }
                headerEmitter.emitHeader(headerName, headerValue, false);
            } else if ((b & 0b11110000) == 0b00010000) {
                first = false;
                //Literal Header Field never indexed
                HttpString headerName = readHeaderName(buffer, 4);
                if (headerName == null) {
                    buffer.position(originalPos);
                    return;
                }
                String headerValue = readHpackString(buffer);
                if (headerValue == null) {
                    if(!moreData) {
                        throw UndertowMessages.MESSAGES.hpackFailed();
                    }
                    buffer.position(originalPos);
                    return;
                }
                headerEmitter.emitHeader(headerName, headerValue, true);
            } else if ((b & 0b11100000) == 0b00100000) {
                if(!first) {
                    throw new HpackException();
                }
                //context update max table size change
                if (!handleMaxMemorySizeChange(buffer, originalPos)) {
                    return;
                }
            } else {
                throw UndertowMessages.MESSAGES.invalidHpackEncoding(b);
            }
        }
        if(!moreData) {
            first = true;
        }
    }
184731
public void handle(final WebSocketSession session, byte[] bytes) {
        if (session == null) {
            throw new NullPointerException("session");
        }
        try {
            JsonRpcMessage msg = jsonRpcControl.receive(bytes);
            if (msg instanceof JsonRpcRequest) {
                handleRequest(session, (JsonRpcRequest) msg);
            } else if (msg instanceof JsonRpcResponse) {
                handleResponse(session, (JsonRpcResponse) msg);
            }
        } catch (Exception e) {
            config.getErrorProcessor().onError(session.getId(), e);
        }
    }
1044089
private boolean readMessage(ByteBuffer buffer) {

    // logger.finest("readMessage: " + bytesRead + " bytes read, remaining=" +
    // remainingBytes.length + ", state=" + state.toString() + ", recurse=" +
    // recurse);

    buffer.position(0);
    int bytesWeHaveBeenReading = 0;

    switch (state) {
      case SERVICELIST:
        StringBuilder builder = new StringBuilder();
        builder.append(buffer.asCharBuffer());
        parseServiceList(builder.toString());
        buffer.position(0);// reset position!
        bytesWeHaveBeenReading = buffer.limit(); // bytesWeHaveBeenreading = all
        // bytes in buffer!
        break;

      case HANDSHAKE:
        if (buffer.limit() >= 6) {
          byte[] dst = new byte[6];
          buffer.position(0);// read from start!
          buffer.get(dst);
          buffer.position(0);// reset position!
          bytesWeHaveBeenReading = 6; // 6 bytes will be removed from buffer
          String handShake = new String(dst);
          if (!handShake.equals("STP/1\n")) {
            close();
            connectionHandler.onException(
                new CommunicationException("Expected STP/1, got: " + handShake));
          }
          setState(State.EMPTY);
          connectionHandler.onHandshake(true);
        }
        break;

      case EMPTY: // read 4 byte header: STP/0
        if (buffer.limit() >= 4) {
          byte[] headerPrefix = new byte[4];
          buffer.get(headerPrefix);
          buffer.position(0);
          bytesWeHaveBeenReading = 4;
          ByteString incomingPrefix = ByteString.copyFrom(headerPrefix);
          if (stpPrefix.equals(incomingPrefix)) {
            setState(State.STP);
            /*
            if(buffer.hasRemaining()){
                buffer.compact();
                readMessage(buffer, buffer.position(), true);
            } else {
                buffer.clear();
            }
            */
          } else {
            close();
            connectionHandler.onException(
                new CommunicationException("Expected empty header"));
          }
        }
        break;

      case STP:
        // Try to read size
        buffer.position(0);
        if (buffer.limit() <= 0) {
          logger.finest("STP: Empty buffer");
          break;
        }
        int messageSize = readRawVarint32(buffer);// read part of buffer
        bytesWeHaveBeenReading = buffer.position();
        buffer.position(0);

        // If we got size, read more, if not just leave it!
        if (buffer.limit() >= bytesWeHaveBeenReading + messageSize) {
          buffer.position(bytesWeHaveBeenReading);

          // Read type and Payload!
          int messageType = buffer.get();
          bytesWeHaveBeenReading += 1;

          byte[] payload = new byte[--messageSize];
          buffer.get(payload);
          buffer.position(0);

          bytesWeHaveBeenReading += messageSize; // 32 bits = 4 bytes :-)

          setState(State.EMPTY);

          try {
            processMessage(messageType, payload);
          } catch (IOException e) {
            close();
            connectionHandler.onException(new CommunicationException(
                "Error while processing the message: " + e.getMessage()));
          }
        } else {
          // 4 + messageSize because of the int at the beginning
          logger.finest(String.format("Tried to read a message and expected %d bytes, but got %d",
                                      (4 + messageSize), buffer.limit()));

          buffer.position(0);
          bytesWeHaveBeenReading = 0;
        }
        break;

    }

    // Pop number of read bytes from
    if (bytesWeHaveBeenReading > 0) {

      // Pop X bytes, and keep message for the rest
      int rest = buffer.limit() - bytesWeHaveBeenReading;
      if (rest <= 0) {
        buffer.clear();
        buffer.limit(0);
      } else {
        byte[] temp = new byte[rest];

        buffer.position(bytesWeHaveBeenReading);
        buffer.get(temp, 0, rest);

        buffer.clear();
        buffer.limit(rest);
        buffer.position(0);
        buffer.put(temp, 0, rest);
        buffer.position(0);// set position back to start!
      }

      logger.finest(String.format("Did read message of %d bytes, new buffer size is %d",
                                  bytesWeHaveBeenReading, buffer.limit()));

      return true; // We did read a message :-)
    } else {
      if (buffer.limit() > 0) {
        logger.finest("Did NOT read message from buffer of size " + buffer.limit());
      } else {
        logger.finest("No messages in empty buffer");
      }
      return false;
    }
  }
376561
public static void process(InputStream inputStream, StreamHandler handler, byte[] buffer) throws IOException {
    LOGGER.trace("process(InputStream, StreamHandler, byte[])");
    LOGGER.debug("buffer size: {} bytes", buffer != null ? buffer.length : "null");
    int read = -1;
    while( (read = inputStream.read(buffer)) != -1 ) {
      LOGGER.debug("{} bytes read from stream", read);
      handler.handleStreamBuffer(buffer, 0, read);
    }
  }
1313326
public function processResponse()
    {
        // In case it comes back as multiple packets (it shouldn't)
        $buffer = new Buffer(implode('', $this->packets_response));

        // Figure out what packet response this is for
        $response_type = $buffer->readString(PHP_EOL);

        // Figure out which packet response this is
        if (empty($response_type) || !array_key_exists($response_type, $this->responses)) {
            throw new Exception(__METHOD__ . " response type '{$response_type}' is not valid");
        }

        // Offload the call
        $results = call_user_func_array([$this, $this->responses[$response_type]], [$buffer]);

        return $results;
    }
1220592
@Override
    public void processMessage(AbsSender absSender, Message message, String[] arguments) {
        execute(absSender, message.getFrom(), message.getChat(), message.getMessageId(), arguments);
    }
239713
async def _process_ack(self, msg):
        
        pub_ack = protocol.PubAck()
        pub_ack.ParseFromString(msg.data)

        # Unblock pending acks queue if required.
        if not self._pending_pub_acks_queue.empty():
            await self._pending_pub_acks_queue.get()

        try:
            cb = self._pub_ack_map[pub_ack.guid]
            await cb(pub_ack)
            del self._pub_ack_map[pub_ack.guid]
        except KeyError:
            # Just skip the pub ack
            return
        except:
            # TODO: Check for protocol error
            return
9176
protected static function process_buffer() {
        // Reset the buffers first in case we get exception from processor.
        $messages = self::$buffer;
        self::$buffer = array();
        $convmessages = self::$convmessagebuffer;
        self::$convmessagebuffer = array();

        foreach ($messages as $message) {
            list($eventdata, $savemessage, $processorlist) = $message;
            self::send_message_to_processors($eventdata, $savemessage, $processorlist);
        }

        foreach ($convmessages as $convmessage) {
            list($eventprocmap, $eventdata, $savemessage) = $convmessage;
            self::send_conversation_message_to_processors($eventprocmap, $eventdata, $savemessage);
        }
    }
908153
private void handleIncomingMessage(SerialMessage incomingMessage) {
		
		logger.debug("Incoming message to process");
		logger.debug(incomingMessage.toString());
		
		switch (incomingMessage.getMessageType()) {
			case Request:
				handleIncomingRequestMessage(incomingMessage);
				break;
			case Response:
				handleIncomingResponseMessage(incomingMessage);
				break;
			default:
				logger.warn("Unsupported incomingMessageType: 0x%02X", incomingMessage.getMessageType());
		}
	}
1030342
@Override
    public int decode ( byte[] buffer, int bufferIndex, int len ) throws SMBProtocolDecodingException {
        int start = bufferIndex;
        if ( len < 24 ) {
            throw new SMBProtocolDecodingException("Invalid resume key");
        }

        this.resumeKey = new byte[24];
        System.arraycopy(buffer, bufferIndex, this.resumeKey, 0, 24);
        bufferIndex += 24;

        SMBUtil.readInt4(buffer, bufferIndex); // contextLength - reserved
        bufferIndex += 4;

        return bufferIndex - start;
    }
91419
public void receiveResponse(byte[] data,SplitDetails[] msg,Integer txID,ReturnCode returnCode) 
	{
		cancelReTransmissionTimer();
		cancelLongtranTimer();
		JainMgcpResponseEvent event = null;

		try 
		{
			event = decodeResponse(data,msg,txID,returnCode);
		} 
		catch (Exception e) 
		{
			logger.error("Could not decode message: ", e);
		}

		event.setTransactionHandle(remoteTID);
		if (this.isProvisional(event.getReturnCode())) 
			resetLongtranTimer();
		
		stack.provider.processMgcpResponseEvent(event, commandEvent);

		if (!this.isProvisional(event.getReturnCode())) 
			this.release();		
	}
544135
private static void decode(InputStream in, StringBuilder result) throws ApiDataException, ApiConnectionException {
        try {
            int len = readLen(in);
            if (len > 0) {
                byte buf[] = new byte[len];
                for (int i = 0; i < len; ++i) {
                    int c = in.read();
                    if (c < 0) {
                        throw new ApiDataException("Truncated data. Expected to read more bytes");
                    }
                    buf[i] = (byte) (c & 0xFF);
                }
                String res = new String(buf, Charset.forName("UTF-8"));
                if (result.length() > 0) {
                    result.append("\n");
                }
                result.append(res);
                decode(in, result);
            }
        } catch (IOException ex) {
            throw new ApiConnectionException(ex.getMessage(), ex);
        }
    }
1059716
@Override
    public int decode ( byte[] buffer, int bufferIndex, int len ) throws SMBProtocolDecodingException {
        int start = bufferIndex;

        // Read total allocation units.
        this.alloc = SMBUtil.readInt8(buffer, bufferIndex);
        bufferIndex += 8;

        // read caller available allocation units
        this.free = SMBUtil.readInt8(buffer, bufferIndex);
        bufferIndex += 8;

        // skip actual free units
        bufferIndex += 8;

        this.sectPerAlloc = SMBUtil.readInt4(buffer, bufferIndex);
        bufferIndex += 4;

        this.bytesPerSect = SMBUtil.readInt4(buffer, bufferIndex);
        bufferIndex += 4;

        return bufferIndex - start;
    }