private DecoderState decodeEnvelope(ByteBuf in) throws IOException {
		// --------------------------------------------------------------------
		// (1) header (EnvelopeEncoder.HEADER_SIZE bytes)
		// --------------------------------------------------------------------
		if (this.currentEnvelope == null) {
			copy(in, this.headerBuffer);

			if (this.headerBuffer.hasRemaining()) {
				return DecoderState.PENDING;
			}
			else {
				this.headerBuffer.flip();

				int magicNum = this.headerBuffer.getInt();
				if (magicNum != OutboundEnvelopeEncoder.MAGIC_NUMBER) {
					throw new IOException("Network stream corrupted: invalid magic" +
							"number in current envelope header.");
				}

				int seqNum = this.headerBuffer.getInt();
				JobID jobId = JobID.fromByteBuffer(this.headerBuffer);
				ChannelID sourceId = ChannelID.fromByteBuffer(this.headerBuffer);

				this.currentEnvelope = new Envelope(seqNum, jobId, sourceId);

				int eventsSize = this.headerBuffer.getInt();
				int bufferSize = this.headerBuffer.getInt();

				this.currentEventsBuffer = eventsSize > 0 ? ByteBuffer.allocate(eventsSize) : null;
				this.currentBufferRequestSize = bufferSize > 0 ? bufferSize : 0;

				this.headerBuffer.clear();
			}
		}

		// --------------------------------------------------------------------
		// (2) events (var length)
		// --------------------------------------------------------------------
		if (this.currentEventsBuffer != null) {
			copy(in, this.currentEventsBuffer);

			if (this.currentEventsBuffer.hasRemaining()) {
				return DecoderState.PENDING;
			}
			else {
				this.currentEventsBuffer.flip();
				this.currentEnvelope.setEventsSerialized(this.currentEventsBuffer);
				this.currentEventsBuffer = null;
			}
		}

		// --------------------------------------------------------------------
		// (3) buffer (var length)
		// --------------------------------------------------------------------
		// (a) request a buffer from OUR pool
		if (this.currentBufferRequestSize > 0) {
			JobID jobId = this.currentEnvelope.getJobID();
			ChannelID sourceId = this.currentEnvelope.getSource();
			Buffer buffer = requestBufferForTarget(jobId, sourceId, this.currentBufferRequestSize);

			if (buffer == null) {
				return DecoderState.NO_BUFFER_AVAILABLE;
			}
			else {
				this.currentEnvelope.setBuffer(buffer);
				this.currentDataBuffer = buffer.getMemorySegment().wrap(0, this.currentBufferRequestSize);
				this.currentBufferRequestSize = 0;
			}
		}

		// (b) copy data to OUR buffer
		if (this.currentDataBuffer != null) {
			copy(in, this.currentDataBuffer);

			if (this.currentDataBuffer.hasRemaining()) {
				return DecoderState.PENDING;
			}
			else {
				this.currentDataBuffer = null;
			}
		}

		// if we made it to this point, we completed the envelope;
		// in the other cases we return early with PENDING or NO_BUFFER_AVAILABLE
		return DecoderState.COMPLETE;
	}
private DecoderState decodeEnvelope ( ByteBuf in ) throws IOException { if ( this . currentEnvelope == null ) { copy ( in , this . headerBuffer ) ; if ( this . headerBuffer . hasRemaining ( ) ) { return DecoderState . PENDING ; } else { this . headerBuffer . flip ( ) ; int magicNum = this . headerBuffer . getInt ( ) ; if ( magicNum != OutboundEnvelopeEncoder . MAGIC_NUMBER ) { throw new IOException ( "Network stream corrupted: invalid magic" + "number in current envelope header." ) ; } int seqNum = this . headerBuffer . getInt ( ) ; JobID jobId = JobID . fromByteBuffer ( this . headerBuffer ) ; ChannelID sourceId = ChannelID . fromByteBuffer ( this . headerBuffer ) ; this . currentEnvelope = new Envelope ( seqNum , jobId , sourceId ) ; int eventsSize = this . headerBuffer . getInt ( ) ; int bufferSize = this . headerBuffer . getInt ( ) ; this . currentEventsBuffer = eventsSize > 0 ? ByteBuffer . allocate ( eventsSize ) : null ; this . currentBufferRequestSize = bufferSize > 0 ? bufferSize : 0 ; this . headerBuffer . clear ( ) ; } } if ( this . currentEventsBuffer != null ) { copy ( in , this . currentEventsBuffer ) ; if ( this . currentEventsBuffer . hasRemaining ( ) ) { return DecoderState . PENDING ; } else { this . currentEventsBuffer . flip ( ) ; this . currentEnvelope . setEventsSerialized ( this . currentEventsBuffer ) ; this . currentEventsBuffer = null ; } } if ( this . currentBufferRequestSize > 0 ) { JobID jobId = this . currentEnvelope . getJobID ( ) ; ChannelID sourceId = this . currentEnvelope . getSource ( ) ; Buffer buffer = requestBufferForTarget ( jobId , sourceId , this . currentBufferRequestSize ) ; if ( buffer == null ) { return DecoderState . NO_BUFFER_AVAILABLE ; } else { this . currentEnvelope . setBuffer ( buffer ) ; this . currentDataBuffer = buffer . getMemorySegment ( ) . wrap ( 0 , this . currentBufferRequestSize ) ; this . currentBufferRequestSize = 0 ; } } if ( this . currentDataBuffer != null ) { copy ( in , this . currentDataBuffer ) ; if ( this . currentDataBuffer . hasRemaining ( ) ) { return DecoderState . PENDING ; } else { this . currentDataBuffer = null ; } } return DecoderState . COMPLETE ; }
--------------------------------------------------------------------