@Override
public void decode ( final IoSession session, final IoBuffer data, final ProtocolDecoderOutput output ) throws Exception
{
    final short magic = data.getShort ();
    final byte version = data.get ();
    final int sequence = data.getInt ();
    final byte commandCode = data.get ();

    if ( magic != 1202 )
    {
        throw new ProtocolCodecException ( String.format ( "Magic code should be 1202 but is %s", magic ) );
    }
    if ( version != 1 )
    {
        throw new ProtocolCodecException ( String.format ( "Version should be %s but is %s", 1, version ) );
    }

    decodeMessage ( sequence, commandCode, data, output );
}
 

private Object decodeData ( final IoBuffer data ) throws ProtocolCodecException
{
    data.order ( ByteOrder.LITTLE_ENDIAN );

    final byte dataType = data.get ();
    switch ( dataType )
    {
        case 0:
            return null;
        case 1:
            return data.get () != 0x00;
        case 2:
            return data.getInt ();
        case 3:
            return data.getLong ();
        case 4:
            return data.getFloat ();
        default:
            throw new ProtocolCodecException ( String.format ( "Data type %02x is unknown", dataType ) );
    }

}
 

private boolean processWriteCommand ( final IoSession session, final IoBuffer data, final ProtocolDecoderOutput out ) throws ProtocolCodecException
{
    final int len = messageLength ( data );
    if ( len < 0 )
    {
        return false;
    }

    final int registerNumber = data.getUnsignedShort ();
    final int operationId = data.getInt ();
    final Variant value = decodeVariant ( session, data );

    out.write ( new WriteCommand ( registerNumber, value, operationId ) );

    return true;
}
 

private boolean processWriteResult ( final IoSession session, final IoBuffer data, final ProtocolDecoderOutput out ) throws ProtocolCodecException
{
    final int len = messageLength ( data );
    if ( len < 0 )
    {
        return false;
    }

    try
    {
        final int operationId = data.getInt ();
        final int errorCode = data.getUnsignedShort ();
        final String errorMessage = decodeString ( session, data );

        out.write ( new WriteResult ( operationId, errorCode, errorMessage ) );
    }
    catch ( final CharacterCodingException e )
    {
        throw new ProtocolCodecException ( e );
    }

    return true;
}
 

private boolean processDataUpdate ( final IoSession session, final IoBuffer data, final ProtocolDecoderOutput out ) throws ProtocolCodecException
{
    final int len = messageLength ( data );
    if ( len < 0 )
    {
        return false;
    }

    final int count = data.getUnsignedShort ();
    final List<DataUpdate.Entry> entries = new ArrayList<DataUpdate.Entry> ( count );
    for ( int i = 0; i < count; i++ )
    {
        entries.add ( decodeDataUpdateEntry ( data, session ) );
    }

    out.write ( new DataUpdate ( entries ) );

    return true;
}
 

private boolean processBrowseAdded ( final IoSession session, final IoBuffer data, final ProtocolDecoderOutput out ) throws ProtocolCodecException
{
    final int len = messageLength ( data );
    if ( len < 0 )
    {
        return false;
    }

    final int count = data.getUnsignedShort ();

    final List<BrowseAdded.Entry> entries = new ArrayList<BrowseAdded.Entry> ( count );

    for ( int i = 0; i < count; i++ )
    {
        entries.add ( decodeBrowserAddEntry ( data, session ) );
    }

    out.write ( new BrowseAdded ( entries ) );

    return true;
}
 

private boolean processHello ( final IoSession session, final IoBuffer data, final ProtocolDecoderOutput out ) throws ProtocolCodecException
{
    final int len = messageLength ( data );
    if ( len < 0 )
    {
        return false;
    }

    final byte version = data.get ();
    if ( version != 0x01 )
    {
        throw new ProtocolCodecException ( String.format ( "Protocol version %s is unsupported", version ) );
    }

    final short nodeId = data.getShort ();
    final EnumSet<Hello.Features> features = data.getEnumSetShort ( Hello.Features.class );

    out.write ( new Hello ( nodeId, features ) );

    return true;
}
 

private void encodeBrowseUpdate ( final IoSession session, final Object message, final IoBuffer data ) throws ProtocolCodecException
{
    // length
    data.putUnsignedShort ( ( (BrowseAdded)message ).getEntries ().size () );

    final CharsetEncoder encoder = Sessions.getCharsetEncoder ( session );

    // data
    for ( final BrowseAdded.Entry entry : ( (BrowseAdded)message ).getEntries () )
    {
        data.putUnsignedShort ( entry.getRegister () );
        data.put ( entry.getDataType ().getDataType () );
        data.putEnumSet ( entry.getFlags () );

        try
        {
            data.putPrefixedString ( entry.getName (), encoder );
            data.putPrefixedString ( entry.getDescription (), encoder );
            data.putPrefixedString ( entry.getUnit (), encoder );
        }
        catch ( final CharacterCodingException e )
        {
            throw new ProtocolCodecException ( e );
        }
    }
}
 

private void encodeProperties ( final IoBuffer data, final Map<String, String> properties ) throws ProtocolCodecException
{
    final CharsetEncoder encoder = this.defaultCharset.newEncoder ();

    data.putUnsignedShort ( properties.size () );
    for ( final Map.Entry<String, String> entry : properties.entrySet () )
    {
        try
        {
            data.putPrefixedString ( entry.getKey (), encoder );
            data.putPrefixedString ( entry.getValue (), encoder );
        }
        catch ( final CharacterCodingException e )
        {
            throw new ProtocolCodecException ( e );
        }
    }
}
 

private byte checkType ( final IoBuffer buffer, final byte expectedType, final boolean allowNull ) throws Exception
{
    final byte type = buffer.get ();

    if ( allowNull && type == TYPE_NULL )
    {
        return type;
    }

    if ( type != expectedType )
    {
        if ( type == 0 && !allowNull )
        {
            throw new ProtocolCodecException ( String.format ( "Failed to decode. Field is transmitted as null but defined as not-null." ) );
        }

        throw new ProtocolCodecException ( String.format ( "Failed to decode string: Expected type %02x, found: %02x", expectedType, type ) );
    }
    return type;
}
 

@Test
public void compareMessages() throws IOException, ProtocolCodecException
{
    IMessage msg1 = new MapMessage("MarketDataSnapshotFullRefresh", "namespace");
    IMessage msg2 = new MapMessage("MarketDataSnapshotFullRefresh", "namespace");

    List<IMessage> list = new ArrayList<>();
    IMessage group = new MapMessage("MDEntries", "namespace");
    group.addField("269", '1');
    group.addField("270", 20.01D);
    group.addField("271", 40.01D);

    list.add(group);
    group = new MapMessage("MDEntries", "namespace");

    group.addField("269", '2');
    group.addField("270", 20.01D);
    group.addField("271", 40.01D);

    list.add(group);
    msg1.addField("MDEntries", list);

    list = new ArrayList<>();
    list.add(group);
    msg2.addField("MDEntries", list);

    ComparatorSettings compSettings = new ComparatorSettings();

    long startTime = System.currentTimeMillis();
    ComparisonResult comparisonResult = MessageComparator.compare(msg1, msg2, compSettings);
    System.out.printf("diff %d\n", System.currentTimeMillis() - startTime);
    System.out.println(comparisonResult);
    Assert.assertEquals(0, ComparisonUtil.getResultCount(comparisonResult, StatusType.FAILED));
}
 

/**
 * Decode a PDU as a slave
 *
 * @param message
 *            the message PDU
 * @return the decoded messages
 * @throws IllegalStateException
 *             if the function code is not supported
 */
public static Object decodeAsSlave ( final Pdu message ) throws ProtocolCodecException
{
    final IoBuffer data = message.getData ();

    final byte functionCode = data.get ();

    switch ( functionCode )
    {
        case Constants.FUNCTION_CODE_READ_COILS:
        case Constants.FUNCTION_CODE_READ_DISCRETE_INPUTS:
        case Constants.FUNCTION_CODE_READ_HOLDING_REGISTERS:
        case Constants.FUNCTION_CODE_READ_INPUT_REGISTERS:
            return new ReadRequest ( message.getTransactionId (), message.getUnitIdentifier (), functionCode, data.getUnsignedShort (), data.getUnsignedShort () );
        case Constants.FUNCTION_CODE_WRITE_SINGLE_COIL:
        case Constants.FUNCTION_CODE_WRITE_SINGLE_REGISTER:
            return new WriteSingleDataRequest ( message.getTransactionId (), message.getUnitIdentifier (), functionCode, data.getUnsignedShort (), readBytes ( data, 2 ) );
        case Constants.FUNCTION_CODE_WRITE_MULTIPLE_COILS:
        case Constants.FUNCTION_CODE_WRITE_MULTIPLE_REGISTERS:
            final int startAddress = data.getUnsignedShort ();
            final int numRegisters = data.getUnsignedShort (); /* number of registers */
            final byte num = data.get ();
            if ( data.remaining () != num )
            {
                throw new ProtocolCodecException ( String.format ( "Wrong amount of data. Announced %s (bytes), found %s (bytes)", num, data.remaining () ) );
            }
            final byte[] b = new byte[data.remaining ()];
            data.get ( b );
            return new WriteMultiDataRequest ( message.getTransactionId (), message.getUnitIdentifier (), functionCode, startAddress, b, numRegisters );
        default:
            throw new IllegalStateException ( String.format ( "Function code %02x is not supported", functionCode ) );
    }
}
 

private void decodeMessage ( final int sequence, final byte commandCode, final IoBuffer data, final ProtocolDecoderOutput output ) throws ProtocolCodecException
{
    switch ( commandCode )
    {
        case 3:
            decodeResponseConfiguration ( data, output, sequence );
            break;
        case 5:
            decodeResponseData ( data, output, sequence );
            break;
    }

}
 

private void decodeResponseData ( final IoBuffer data, final ProtocolDecoderOutput output, final int sequence ) throws ProtocolCodecException
{
    final byte nin = data.get ();

    final Object[] vars = new Object[nin];

    for ( int i = 0; i < nin; i++ )
    {
        vars[i] = decodeData ( data );
    }

    final DataMessage msg = new DataMessage ( sequence, CommandCode.RESPOND_DATA, vars );
    output.write ( msg );
}
 

/**
 * Decode a variant from the stream
 *
 * @return a decoded variant or <code>null</code> if the data type was
 *         {@link DataType#DEAD}
 * @throws ProtocolCodecException
 */
private Variant decodeVariant ( final IoSession session, final IoBuffer data ) throws ProtocolCodecException
{
    final byte b = data.get ();
    final DataType dataType = DataType.fromByte ( b );
    if ( dataType == null )
    {
        throw new ProtocolCodecException ( String.format ( "Data type %02x is unkown", b ) );
    }

    switch ( dataType )
    {
        case DEAD:
            return null;
        case NULL:
            return Variant.NULL;
        case BOOLEAN:
            return Variant.valueOf ( data.get () != 0x00 );
        case INT32:
            return Variant.valueOf ( data.getInt () );
        case INT64:
            return Variant.valueOf ( data.getInt () );
        case DOUBLE:
            return Variant.valueOf ( data.getDouble () );
        case STRING:
            try
            {
                return Variant.valueOf ( decodeString ( session, data ) );
            }
            catch ( final CharacterCodingException e )
            {
                throw new ProtocolCodecException ( e );
            }
        default:
            throw new ProtocolCodecException ( String.format ( "Data type %s is unkown", b ) );
    }
}
 

private DataUpdate.Entry decodeDataUpdateEntry ( final IoBuffer data, final IoSession session ) throws ProtocolCodecException
{
    final int register = data.getUnsignedShort ();
    final byte missedUpdates = data.get ();
    final long timestamp = data.getLong ();
    final Set<DataUpdate.State> states = data.getEnumSetShort ( DataUpdate.State.class );

    final Variant value = decodeVariant ( session, data );

    return new DataUpdate.Entry ( register, value, timestamp, states, missedUpdates );
}
 

private BrowseAdded.Entry decodeBrowserAddEntry ( final IoBuffer data, final IoSession session ) throws ProtocolCodecException
{
    final short register = (short)data.getUnsignedShort ();
    // FIXME: validate if short works

    final byte b = data.get ();
    final DataType dataType = DataType.fromByte ( b );

    if ( dataType == null )
    {
        throw new ProtocolCodecException ( String.format ( "Data type %s is unkown", b ) );
    }

    final Set<BrowseAdded.Entry.Flags> flags = data.getEnumSet ( BrowseAdded.Entry.Flags.class );

    final CharsetDecoder decoder = Sessions.getCharsetDecoder ( session );

    try
    {
        final String name = data.getPrefixedString ( decoder );
        final String description = data.getPrefixedString ( decoder );
        final String unit = data.getPrefixedString ( decoder );
        return new BrowseAdded.Entry ( register, name, description, unit, dataType, flags );
    }
    catch ( final CharacterCodingException e )
    {
        throw new ProtocolCodecException ( e );
    }

}
 

private void encodeEntry ( final IoSession session, final IoBuffer data, final Entry entry ) throws ProtocolCodecException
{
    data.putUnsignedShort ( entry.getRegister () );
    data.put ( entry.getMissedUpdates () );
    data.putLong ( entry.getTimestamp () );
    data.putEnumSetShort ( entry.getStates () );

    // put payload
    encodeVariant ( session, data, entry.getValue () );
}
 

@Override
public void encode(IoSession session, Object message, ProtocolEncoderOutput out) throws ProtocolCodecException {
    // get the connection from the session
    String sessionId = (String) session.getAttribute(RTMPConnection.RTMP_SESSION_ID);
    log.trace("Session id: {}", sessionId);
    RTMPConnection conn = (RTMPConnection) RTMPConnManager.getInstance().getConnectionBySessionId(sessionId);
    if (conn != null) {
        Red5.setConnectionLocal(conn);
        final Semaphore lock = conn.getEncoderLock();
        try {
            // acquire the encoder lock
            lock.acquire();
            // get the buffer
            final IoBuffer buf = message instanceof IoBuffer ? (IoBuffer) message : getEncoder().encode(message);
            if (buf != null) {
                if (log.isTraceEnabled()) {
                    log.trace("Writing output data: {}", Hex.encodeHexString(buf.array()));
                }
                out.write(buf);
            } else {
                log.trace("Response buffer was null after encoding");
            }
        } catch (Exception ex) {
            log.error("Exception during encode", ex);
        } finally {
            lock.release();
            Red5.setConnectionLocal(null);
        }
    } else {
        log.debug("Connection is no longer available for encoding, may have been closed already");
    }
}
 

@Test
public void testAnyRepeatingGroup() throws IOException, ProtocolCodecException
{
    IMessage message = new MapMessage("Message", "namespace");
    IMessage filter = new MapMessage("Message", "namespace");

    List<IMessage> list = new ArrayList<>();
    IMessage group = new MapMessage("Group", "namespace");
    group.addField("269", '1');
    group.addField("270", 20.01D);
    group.addField("271", 40.01D);

    list.add(group);
    list.add(group);

    message.addField("Group", list);
    filter.addField("Group", StaticUtil.notNullFilter(0, null));

    ComparatorSettings compSettings = new ComparatorSettings();
    MetaContainer metaContainer = new MetaContainer();

    compSettings.setMetaContainer(metaContainer);

    ComparisonResult comparisonResult = MessageComparator.compare(message, filter, compSettings);
    System.out.println(comparisonResult);
    Assert.assertEquals(0, ComparisonUtil.getResultCount(comparisonResult, StatusType.FAILED));
    Assert.assertEquals(8, ComparisonUtil.getResultCount(comparisonResult, StatusType.NA));
    Assert.assertEquals(1, ComparisonUtil.getResultCount(comparisonResult, StatusType.PASSED));

    metaContainer.setFailUnexpected("Y");
    comparisonResult = MessageComparator.compare(message, filter, compSettings);
    System.out.println(comparisonResult);
    Assert.assertEquals(0, ComparisonUtil.getResultCount(comparisonResult, StatusType.FAILED));
    Assert.assertEquals(8, ComparisonUtil.getResultCount(comparisonResult, StatusType.NA));
    Assert.assertEquals(1, ComparisonUtil.getResultCount(comparisonResult, StatusType.PASSED));

    metaContainer.setFailUnexpected("A");
    comparisonResult = MessageComparator.compare(message, filter, compSettings);
    System.out.println(comparisonResult);
    Assert.assertEquals(0, ComparisonUtil.getResultCount(comparisonResult, StatusType.FAILED));
    Assert.assertEquals(8, ComparisonUtil.getResultCount(comparisonResult, StatusType.NA));
    Assert.assertEquals(1, ComparisonUtil.getResultCount(comparisonResult, StatusType.PASSED));
}
 

@Test
public void testUnchecked() throws IOException, ProtocolCodecException
{
    IMessage message = new MapMessage("MarketDataSnapshotFullRefresh", "namespace");
    IMessage filter = new MapMessage("MarketDataSnapshotFullRefresh", "namespace");


    IMessage subMsg1 = new MapMessage("List", "namespace");
    subMsg1.addField("LP", '1');
    subMsg1.addField("LN", '2');
    subMsg1.addField("LF", '3');

    IMessage subMsg2 = new MapMessage("List", "namespace");
    subMsg2.addField("LP", '1');
    subMsg2.addField("LF", '4');

    List<IMessage> list1 = new ArrayList<>();
    List<IMessage> list2 = new ArrayList<>();

    list1.add(subMsg1);
    list1.add(subMsg1);

    list2.add(subMsg2);
    list2.add(subMsg2);

    subMsg1 = new MapMessage("Message", "namespace");
    subMsg1.addField("MP", '1');
    subMsg1.addField("MN", '2');
    subMsg1.addField("MF", '3');

    subMsg2 = new MapMessage("Message", "namespace");
    subMsg2.addField("MP", '1');
    subMsg2.addField("MF", '4');

    message.addField("FieldL", list1);
    message.addField("FieldM", subMsg1);
    message.addField("P", '1');
    message.addField("N", '2');
    message.addField("F", '3');
    message.addField("Sub", message.cloneMessage());

    filter.addField("FieldL", list2);
    filter.addField("FieldM", subMsg2);
    filter.addField("P", '1');
    filter.addField("F", '4');
    filter.addField("Sub", filter.cloneMessage());

    MetaContainer metaContainer = new MetaContainer();
    metaContainer.setFailUnexpected("A");

    ComparatorSettings compSettings = new ComparatorSettings();
    Set<String> set = new HashSet<>();
    set.add("N");
    set.add("FieldL");
    set.add("FieldM");
    set.add("List"); // No effect
    set.add("Message"); // No effect
    set.add("MarketDataSnapshotFullRefresh"); // No effect
    compSettings.setUncheckedFields(set);
    compSettings.setMetaContainer(metaContainer);

    ComparisonResult comparisonResult = MessageComparator.compare(message, filter, compSettings);
    System.out.println(comparisonResult);
    Assert.assertEquals(8, ComparisonUtil.getResultCount(comparisonResult, StatusType.FAILED));
    Assert.assertEquals(8, ComparisonUtil.getResultCount(comparisonResult, StatusType.NA));
    Assert.assertEquals(8, ComparisonUtil.getResultCount(comparisonResult, StatusType.PASSED));
}
 

@Test
public void compareLargeMessages() throws IOException, ProtocolCodecException
   {
    int count = 500;

    IMessage msg1 = new MapMessage("MarketDataSnapshotFullRefresh", "namespace");

    IMessage msgHeader = new MapMessage("Header", "namespace");
    msgHeader.addField("8", "FIXT.1.1");
    msgHeader.addField("9", "11789");
    msgHeader.addField("35", "W");
       msgHeader.addField("49", "Sender");
    msgHeader.addField("56", "PVMEMD");
    msgHeader.addField("34", "251");
    msgHeader.addField("52", "20150429-15:30:19.900");

    msg1.addField("header", msgHeader);
    msg1.addField("262", "1430321440762");
    msg1.addField("55", "[N/A]");
    msg1.addField("48", "EURv3MEURIB:(10Yv20Y)");
    msg1.addField("22", "101");
    msg1.addField("268", count);

    IMessage msg2 = msg1.cloneMessage();

    IMessage group = new MapMessage("MDEntries", "namespace");
       group.addField("269", '2');
       group.addField("270", 40.01D);
       group.addField("271", 20.01D);
       group.addField("273", DateTimeUtility.nowLocalDateTime());
       group.addField("274", DateTimeUtility.nowLocalDate());
       group.addField("275", DateTimeUtility.nowLocalTime());
       group.addField("336", 7);
       group.addField("326", 17);

       List<IMessage> list = new ArrayList<>();
    for (int i = 0; i < count; i++) {
        list.add(group);
       }
    msg1.addField("MDEntries", list);

    List<IMessage> list2 = new ArrayList<>();
    list2.add(group);
    msg2.addField("MDEntries", list2);

       ComparatorSettings compSettings = new ComparatorSettings();

       long startTime = System.currentTimeMillis();
       ComparisonResult comparisonResult = MessageComparator.compare(msg1, msg2, compSettings);
       System.out.printf("diff %d\n", System.currentTimeMillis() - startTime);
       Assert.assertEquals(0, ComparisonUtil.getResultCount(comparisonResult, StatusType.FAILED));
   }
 

@Override
protected boolean doDecode ( final IoSession session, final IoBuffer data, final ProtocolDecoderOutput out ) throws Exception
{
    byte marker1;
    byte marker2;

    do
    {
        if ( data.remaining () < 2 ) // we may only eat the start when there could be a packet after it
        {
            return false;
        }

        // peek marker
        marker1 = data.get ( data.position () + 0 );
        marker2 = data.get ( data.position () + 1 );

        // TODO: re-think the idea of just skipping

        if ( marker1 != 0x12 || marker2 != 0x02 )
        {
            data.skip ( 2 ); // eat garbage
        }
    } while ( marker1 != 0x12 || marker2 != 0x02 );

    if ( data.remaining () < 3 )
    {
        return false;
    }

    data.order ( Sessions.isLittleEndian ( session ) ? ByteOrder.LITTLE_ENDIAN : ByteOrder.BIG_ENDIAN );

    final byte command = data.get ( data.position () + 2 );
    switch ( command )
    {
        case Messages.MC_HELLO:
            return processHello ( session, data, out );
        case Messages.MC_WELCOME:
            return processWelcome ( session, data, out );
        case Messages.MC_READ_ALL:
            out.write ( new ReadAll () );
            return true;
        case Messages.MC_DATA_UPDATE:
            return processDataUpdate ( session, data, out );
        case Messages.MC_START_BROWSE:
            out.write ( new SubscribeBrowse () );
            return true;
        case Messages.MC_STOP_BROWSE:
            out.write ( new UnsubscribeBrowse () );
            return true;
        case Messages.MC_NS_ADDED:
            return processBrowseAdded ( session, data, out );
        case Messages.MC_WRITE_COMMAND:
            return processWriteCommand ( session, data, out );
        case Messages.MC_WRITE_RESULT:
            return processWriteResult ( session, data, out );
    }

    throw new ProtocolCodecException ( String.format ( "Message code %02x is unkown", command ) );
}
 

private boolean processWelcome ( final IoSession session, final IoBuffer data, final ProtocolDecoderOutput out ) throws ProtocolCodecException
{
    final int len = messageLength ( data );
    if ( len < 0 )
    {
        return false;
    }

    final EnumSet<Welcome.Features> features = data.getEnumSetShort ( Welcome.Features.class );

    logger.debug ( "Features: {}", features );

    final CharsetDecoder decoder = this.defaultCharset.newDecoder ();

    final int count = data.getUnsignedShort ();
    final Map<String, String> properties = new HashMap<String, String> ( count );

    for ( int i = 0; i < count; i++ )
    {
        try
        {
            final String key = data.getPrefixedString ( decoder );
            final String value = data.getPrefixedString ( decoder );

            properties.put ( key, value );
        }
        catch ( final CharacterCodingException e )
        {
            throw new ProtocolCodecException ( e );
        }
    }

    if ( features.contains ( Welcome.Features.LITTLE_ENDIAN ) )
    {
        logger.info ( "Setting little endian" );
        Sessions.setLittleEndian ( session );
    }

    out.write ( new Welcome ( features, properties ) );

    return true;
}
 

@Override
public void encode ( final IoSession session, final Object message, final ProtocolEncoderOutput out ) throws Exception
{
    IoBuffer data = null;
    if ( message instanceof Hello )
    {
        data = createMessage ( session, Messages.MC_HELLO, true );

        data.put ( (byte)0x01 ); // version

        data.putShort ( ( (Hello)message ).getNodeId () );
        data.putEnumSetShort ( ( (Hello)message ).getFeatures () );

        data.putUnsignedShort ( 3, data.position () - 3 ); // fill length
    }
    else if ( message instanceof Welcome )
    {
        data = createMessage ( session, Messages.MC_WELCOME, true );

        // put bit set
        data.putEnumSetShort ( ( (Welcome)message ).getFeatures () );
        encodeProperties ( data, ( (Welcome)message ).getProperties () );

        data.putUnsignedShort ( 3, data.position () - 3 ); // fill length
    }
    else if ( message instanceof ReadAll )
    {
        data = createMessage ( session, Messages.MC_READ_ALL, false );

    }
    else if ( message instanceof DataUpdate )
    {
        data = createMessage ( session, Messages.MC_DATA_UPDATE, true );

        data.putUnsignedShort ( ( (DataUpdate)message ).getEntries ().size () );
        // put values
        for ( final DataUpdate.Entry entry : ( (DataUpdate)message ).getEntries () )
        {
            encodeEntry ( session, data, entry );
        }

        data.putUnsignedShort ( 3, data.position () - 3 ); // fill length
    }
    else if ( message instanceof SubscribeBrowse )
    {
        data = createMessage ( session, Messages.MC_START_BROWSE, false );
    }
    else if ( message instanceof UnsubscribeBrowse )
    {
        data = createMessage ( session, Messages.MC_STOP_BROWSE, false );
    }
    else if ( message instanceof BrowseAdded )
    {
        data = createMessage ( session, Messages.MC_NS_ADDED, true );

        // put browse update
        encodeBrowseUpdate ( session, message, data );

        data.putUnsignedShort ( 3, data.position () - 3 ); // fill length
    }
    else if ( message instanceof WriteCommand )
    {
        data = createMessage ( session, Messages.MC_WRITE_COMMAND, true );

        data.putUnsignedShort ( ( (WriteCommand)message ).getRegisterNumber () );
        data.putInt ( ( (WriteCommand)message ).getOperationId () );
        encodeVariant ( session, data, ( (WriteCommand)message ).getValue () );

        data.putUnsignedShort ( 3, data.position () - 3 ); // fill length
    }
    else if ( message instanceof WriteResult )
    {
        data = createMessage ( session, Messages.MC_WRITE_RESULT, true );

        data.putInt ( ( (WriteResult)message ).getOperationId () );
        data.putUnsignedShort ( ( (WriteResult)message ).getErrorCode () );

        data.putPrefixedString ( ( (WriteResult)message ).getErrorMessage (), Sessions.getCharsetEncoder ( session ) );

        data.putUnsignedShort ( 3, data.position () - 3 ); // fill length
    }

    if ( data != null )
    {
        data.flip ();
        out.write ( data );
    }
    else
    {
        throw new ProtocolCodecException ( String.format ( "Message %s is not supported", message.getClass ().getName () ) );
    }
}
 

private void encodeVariant ( final IoSession session, final IoBuffer data, final Variant value ) throws ProtocolCodecException
{
    if ( value == null )
    {
        data.put ( DataType.DEAD.getDataType () ); // dead
    }
    else if ( value.isNull () )
    {
        data.put ( DataType.NULL.getDataType () );
    }
    else if ( value.isBoolean () )
    {
        data.put ( DataType.BOOLEAN.getDataType () );
        data.put ( (byte) ( value.asBoolean () ? 0xFF : 0x00 ) );
    }
    else if ( value.isInteger () )
    {
        data.put ( DataType.INT32.getDataType () );
        data.putInt ( value.asInteger ( null ) );
    }
    else if ( value.isLong () )
    {
        data.put ( DataType.INT64.getDataType () );
        data.putLong ( value.asLong ( null ) );
    }
    else if ( value.isDouble () )
    {
        data.put ( DataType.DOUBLE.getDataType () );
        data.putDouble ( value.asDouble ( null ) );
    }
    else if ( value.isString () )
    {
        data.put ( DataType.STRING.getDataType () );
        try
        {
            data.putPrefixedString ( value.asString ( null ), Sessions.getCharsetEncoder ( session ) );
        }
        catch ( final CharacterCodingException e )
        {
            throw new ProtocolCodecException ( e );
        }
    }
}
 

/** {@inheritDoc} */
public void decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws ProtocolCodecException {
    // get the connection from the session
    String sessionId = (String) session.getAttribute(RTMPConnection.RTMP_SESSION_ID);
    log.trace("Session id: {}", sessionId);
    // connection verification routine
    @SuppressWarnings("unchecked")
    IConnectionManager<RTMPConnection> connManager = (IConnectionManager<RTMPConnection>) ((WeakReference<?>) session.getAttribute(RTMPConnection.RTMP_CONN_MANAGER)).get();
    RTMPConnection conn = (RTMPConnection) connManager.getConnectionBySessionId(sessionId);
    RTMPConnection connLocal = (RTMPConnection) Red5.getConnectionLocal();
    if (connLocal == null || !conn.getSessionId().equals(connLocal.getSessionId())) {
        if (log.isDebugEnabled() && connLocal != null) {
            log.debug("Connection local didn't match session");
        }
    }
    if (conn != null) {
        // set the connection to local if its referred to by this session
        Red5.setConnectionLocal(conn);
        // copy data range from incoming
        if (log.isTraceEnabled()) {
            log.trace("Incoming: in.position {}, in.limit {}, in.remaining {}", new Object[] { in.position(), in.limit(), in.remaining() });
        }
        byte[] arr = new byte[in.remaining()];
        in.get(arr);
        // create a buffer and store it on the session
        IoBuffer buf = (IoBuffer) session.getAttribute("buffer");
        if (buf == null) {
            buf = IoBuffer.allocate(arr.length);
            buf.setAutoExpand(true);
            session.setAttribute("buffer", buf);
        }
        // copy incoming into buffer
        buf.put(arr);
        // flip so we can read
        buf.flip();
        if (log.isTraceEnabled()) {
            //log.trace("Buffer before: {}", Hex.encodeHexString(arr));
            log.trace("Buffers info before: buf.position {}, buf.limit {}, buf.remaining {}", new Object[] { buf.position(), buf.limit(), buf.remaining() });
        }
        // get the connections decoder lock
        final Semaphore lock = conn.getDecoderLock();
        try {
            // acquire the decoder lock
            lock.acquire();
            // construct any objects from the decoded bugger
            List<?> objects = decoder.decodeBuffer(conn, buf);
            log.trace("Decoded: {}", objects);
            if (objects != null) {
                int writeCount = 0;
                for (Object object : objects) {
                    out.write(object);
                    writeCount++;
                }
                log.trace("Wrote {} objects", writeCount);
            }
        } catch (Exception e) {
            log.error("Error during decode", e);
        } finally {
            lock.release();
            // clear local
            Red5.setConnectionLocal(null);
        }
        if (log.isTraceEnabled()) {
            //log.trace("Buffer after: {}", Hex.encodeHexString(Arrays.copyOfRange(buf.array(), buf.position(), buf.limit())));
            log.trace("Buffers info after: buf.position {}, buf.limit {}, buf.remaining {}", new Object[] { buf.position(), buf.limit(), buf.remaining() });
        }
    } else {
        log.debug("Closing and skipping decode for unregistered connection: {}", sessionId);
        session.closeNow();
        log.debug("Session closing: {} reading: {} writing: {}", session.isClosing(), session.isReadSuspended(), session.isWriteSuspended());
    }
}
 

@Override
public void decode(IoSession session, IoBuffer in, ProtocolDecoderOutput out) throws ProtocolCodecException {
    log.trace("decode buffer position: {}", in.position());
    // get the connection from the session
    String sessionId = (String) session.getAttribute(RTMPConnection.RTMP_SESSION_ID);
    log.trace("Session id: {}", sessionId);
    RTMPConnection conn = (RTMPConnection) RTMPConnManager.getInstance().getConnectionBySessionId(sessionId);
    Red5.setConnectionLocal(conn);
    byte[] arr = new byte[in.remaining()];
    in.get(arr);
    // create a buffer and store it on the session
    IoBuffer buf = (IoBuffer) session.getAttribute("buffer");
    if (buf == null) {
        buf = IoBuffer.allocate(arr.length);
        buf.setAutoExpand(true);
        session.setAttribute("buffer", buf);
    }
    // copy incoming into buffer
    buf.put(arr);
    // flip so we can read
    buf.flip();
    final Semaphore lock = conn.getDecoderLock();
    try {
        // acquire the decoder lock
        lock.acquire();
        // construct any objects from the decoded buffer
        List<?> objects = getDecoder().decodeBuffer(conn, buf);
        log.trace("Decoded: {}", objects);
        if (objects != null) {
            for (Object object : objects) {
                log.trace("Writing {} to decoder output: {}", object, out);
                out.write(object);
            }
        }
        log.trace("Input buffer position: {}", in.position());
    } catch (Exception e) {
        log.error("Error during decode", e);
    } finally {
        lock.release();
        Red5.setConnectionLocal(null);
    }
}