/**
 * Copyright 2012 Ronen Hamias, Anton Kharenko
 *
 * Licensed 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.socketio.netty.serialization;

import java.io.IOException;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufProcessor;
import io.netty.util.CharsetUtil;
import org.socketio.netty.packets.Packet;
import org.socketio.netty.packets.PacketType;

public final class PacketDecoder {

	private static final byte DELIMITER = (byte) ':';
	private static final ByteBufProcessor packetDelimiterFinder =  new ByteBufProcessor() {
		@Override
		public boolean process(byte value) throws Exception {
			return value != DELIMITER;
		}
	};

	/**
	 * Don't let anyone instantiate this class.
	 */
	private PacketDecoder() {
	}

	public static Packet decodePacket(final ByteBuf payload) throws IOException {
		int payloadSize = payload.readableBytes();

		// Decode packet type
		int typeDelimiterIndex = payload.forEachByte(packetDelimiterFinder);
		if (typeDelimiterIndex == -1) {
			return Packet.NULL_INSTANCE;
		}
		ByteBuf typeBytes = payload.slice(0, typeDelimiterIndex);
		String typeString = typeBytes.toString(CharsetUtil.UTF_8);
		int typeId = Integer.valueOf(typeString);
		PacketType type = PacketType.valueOf(typeId);

		// Skip message id
		int messageIdDelimiterIndex = payload.forEachByte(typeDelimiterIndex + 1, payloadSize - typeDelimiterIndex - 1, packetDelimiterFinder);
		if (messageIdDelimiterIndex == -1) {
			return Packet.NULL_INSTANCE;
		}

		// Skip endpoint
		int endpointDelimiterIndex = payload.forEachByte(messageIdDelimiterIndex + 1, payloadSize - messageIdDelimiterIndex - 1, packetDelimiterFinder);

		// Create instance of packet
		Packet packet = new Packet(type);

		// Decode data
		if (endpointDelimiterIndex != -1) {
			ByteBuf data = payload.copy(endpointDelimiterIndex + 1, payloadSize - endpointDelimiterIndex - 1);
			packet.setData(data);
		}

		return packet;
	}

}

@@ -2239,59 +2239,159 @@
 a%0A%09%09
-if (endpointDelimiterIndex != -1) %7B%0A%09%09%09ByteBuf data
+boolean messagingType = type == PacketType.MESSAGE %7C%7C type == PacketType.JSON;%0A%09%09if (endpointDelimiterIndex != -1 && messagingType) %7B%0A%09%09%09int dataLength
  = p
@@ -2400,14 +2400,15 @@
 load
-.copy(
+Size - 
 endp
@@ -2422,36 +2422,80 @@
 imiterIndex 
-+ 1,
+- 1;%0A%09%09%09if (dataLength %3E 0) %7B%0A%09%09%09%09ByteBuf data =
  payload
 Size - endpo
@@ -2474,39 +2474,38 @@
 f data = payload
-Size - 
+.copy(
 endpointDelimite
@@ -2515,14 +2515,27 @@
 dex 
--
++
  1
+, dataLength
 );%0A
+%09
 %09%09%09p
@@ -2555,16 +2555,21 @@
 (data);%0A
+%09%09%09%7D%0A
 %09%09%7D%0A%0A%09%09r

Fix memory leak