/*
 * Copyright (c) 2018, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 *  WSO2 Inc. licenses this file to you under the Apache License,
 *  Version 2.0 (the "License"); you may not use this file except
 *  in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied. See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 */

package org.wso2.carbon.stream.processor.core.ha.transport.handlers;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;
import java.util.concurrent.BlockingQueue;

/**
 * Byte to message decoder.
 */
public class MessageDecoder extends ByteToMessageDecoder {
    private BlockingQueue<ByteBuf> byteBufferQueue;

    public MessageDecoder(BlockingQueue<ByteBuf> byteBufferQueue){
        this.byteBufferQueue = byteBufferQueue;
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        if (in.readableBytes() < 5) {
            return;
        }
        int protocol = in.readByte();
        int messageSize = in.readInt();
        if (protocol != 2 || messageSize > in.readableBytes()) {
            in.resetReaderIndex();
            return;
        }
        byteBufferQueue.offer(in);
        in.markReaderIndex();
    }
}

@@ -881,16 +881,288 @@
 ecoder;%0A
+import org.apache.log4j.Logger;%0Aimport org.wso2.carbon.stream.processor.core.ha.util.HAConstants;%0Aimport org.wso2.carbon.stream.processor.core.util.BinaryMessageConverterUtil;%0A%0Aimport java.io.UnsupportedEncodingException;%0Aimport java.nio.ByteBuffer;%0Aimport java.util.Date;
 %0Aimport 
@@ -1368,16 +1368,179 @@
 rQueue;%0A
+    private static long startTime;%0A    private static long endTime;%0A    static int count = 0;%0A    static final Logger log = Logger.getLogger(MessageDecoder.class);
 %0A    pub
@@ -2064,16 +2064,1200 @@
 er(in);%0A
+        try %7B%0A%0A            int channelIdSize = in.readInt();%0A            String channelId = BinaryMessageConverterUtil.getString(in, channelIdSize);%0A%0A            int dataLength = in.readInt();%0A            byte%5B%5D bytes = new byte%5BdataLength%5D;%0A            in.readBytes(bytes);%0A            //LOG.info(%22Message Received : %22 + new String(bytes));%0A%0A            if (channelId.equals(HAConstants.CHANNEL_ID_MESSAGE)) %7B%0A                ByteBuffer messageBuffer = ByteBuffer.wrap(bytes);%0A                count++;%0A                if (count %25 10000 == 0) %7B%0A                    log.info(%22COUNT %22 + count);%0A                    endTime = new Date().getTime();%0A                    //LOG.info(%22TIME START    %22 + time + %22    TIMES   %22 + timeE);%0A                    log.info(%22Server TPS:   %22 + (((10000 * 1000) / (endTime - startTime))));%0A                    startTime = new Date().getTime();%0A                %7D%0A%0A            %7D else if (channelId.equals(HAConstants.CHANNEL_ID_CONTROL_MESSAGE)) %7B%0A%0A                log.info(%22Control Message Received : %22 + new String(bytes));%0A            %7D%0A        %7D catch (UnsupportedEncodingException e) %7B%0A            e.printStackTrace();%0A        %7D finally %7B%0A            
         
@@ -3274,20 +3274,31 @@
 derIndex();%0A
+        %7D%0A%0A
     %7D%0A%7D%0A

Add tps calculation