Nikita
13 years ago
8 changed files with 314 additions and 79 deletions
-
6pom.xml
-
84src/main/java/com/corundumstudio/socketio/PacketHandler.java
-
12src/main/java/com/corundumstudio/socketio/SocketIOPipelineFactory.java
-
25src/main/java/com/corundumstudio/socketio/messages/PacketsMessage.java
-
107src/main/java/com/corundumstudio/socketio/transport/WebSocketTransport.java
-
6src/main/java/com/corundumstudio/socketio/transport/XHRPollingClient.java
-
30src/main/java/com/corundumstudio/socketio/transport/XHRPollingTransport.java
-
123src/test/java/com/corundumstudio/socketio/PacketHandlerTest.java
@ -0,0 +1,84 @@ |
|||
package com.corundumstudio.socketio; |
|||
|
|||
import java.io.IOException; |
|||
|
|||
import org.jboss.netty.buffer.ChannelBuffer; |
|||
import org.jboss.netty.channel.ChannelHandlerContext; |
|||
import org.jboss.netty.channel.MessageEvent; |
|||
import org.jboss.netty.channel.SimpleChannelUpstreamHandler; |
|||
import org.jboss.netty.channel.ChannelHandler.Sharable; |
|||
import org.jboss.netty.util.CharsetUtil; |
|||
import org.slf4j.Logger; |
|||
import org.slf4j.LoggerFactory; |
|||
|
|||
import com.corundumstudio.socketio.messages.PacketsMessage; |
|||
import com.corundumstudio.socketio.parser.Decoder; |
|||
import com.corundumstudio.socketio.parser.Packet; |
|||
|
|||
@Sharable |
|||
public class PacketHandler extends SimpleChannelUpstreamHandler { |
|||
|
|||
private final Logger log = LoggerFactory.getLogger(getClass()); |
|||
|
|||
private final PacketListener packetListener; |
|||
private final Decoder decoder; |
|||
|
|||
public PacketHandler(PacketListener packetListener, Decoder decoder) { |
|||
super(); |
|||
this.packetListener = packetListener; |
|||
this.decoder = decoder; |
|||
} |
|||
|
|||
@Override |
|||
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) |
|||
throws Exception { |
|||
Object msg = e.getMessage(); |
|||
if (msg instanceof PacketsMessage) { |
|||
PacketsMessage message = (PacketsMessage) msg; |
|||
ChannelBuffer content = message.getContent(); |
|||
|
|||
if (log.isTraceEnabled()) { |
|||
log.trace("In message: {} sessionId: {}", new Object[] {content.toString(CharsetUtil.UTF_8), message.getClient().getSessionId()}); |
|||
} |
|||
while (content.readable()) { |
|||
Packet packet = decode(content); |
|||
packetListener.onPacket(packet, message.getClient()); |
|||
} |
|||
} else { |
|||
ctx.sendUpstream(e); |
|||
} |
|||
} |
|||
|
|||
// TODO use ForkJoin |
|||
private Packet decode(ChannelBuffer buffer) throws IOException { |
|||
char delimiter = getChar(buffer, buffer.readerIndex()); |
|||
if (delimiter == Packet.DELIMITER) { |
|||
StringBuilder length = new StringBuilder(4); |
|||
for (int i = buffer.readerIndex() + 2 + 1; i < buffer.readerIndex() + buffer.readableBytes(); i++) { |
|||
if (getChar(buffer, i) == Packet.DELIMITER) { |
|||
break; |
|||
} else { |
|||
length.append((char)buffer.getUnsignedByte(i)); |
|||
} |
|||
} |
|||
Integer len = Integer.valueOf(length.toString()); |
|||
|
|||
int startIndex = buffer.readerIndex() + 3 + length.length() + 3; |
|||
ChannelBuffer frame = buffer.slice(startIndex, len); |
|||
Packet packet = decoder.decodePacket(frame.toString(CharsetUtil.UTF_8)); |
|||
buffer.readerIndex(startIndex + len); |
|||
return packet; |
|||
} else { |
|||
Packet packet = decoder.decodePacket(buffer.toString(CharsetUtil.UTF_8)); |
|||
buffer.readerIndex(buffer.readableBytes()); |
|||
return packet; |
|||
} |
|||
} |
|||
|
|||
// TODO refactor it |
|||
private char getChar(ChannelBuffer buffer, int index) { |
|||
byte[] bytes = {buffer.getByte(index), buffer.getByte(index + 1)}; |
|||
return new String(bytes).charAt(0); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,25 @@ |
|||
package com.corundumstudio.socketio.messages; |
|||
|
|||
import org.jboss.netty.buffer.ChannelBuffer; |
|||
|
|||
import com.corundumstudio.socketio.SocketIOClient; |
|||
|
|||
public class PacketsMessage { |
|||
|
|||
private final SocketIOClient client; |
|||
private final ChannelBuffer content; |
|||
|
|||
public PacketsMessage(SocketIOClient client, ChannelBuffer content) { |
|||
this.client = client; |
|||
this.content = content; |
|||
} |
|||
|
|||
public SocketIOClient getClient() { |
|||
return client; |
|||
} |
|||
|
|||
public ChannelBuffer getContent() { |
|||
return content; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,123 @@ |
|||
package com.corundumstudio.socketio; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Collections; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
import junit.framework.Assert; |
|||
import mockit.Mocked; |
|||
|
|||
import org.codehaus.jackson.map.ObjectMapper; |
|||
import org.jboss.netty.buffer.ChannelBuffer; |
|||
import org.jboss.netty.buffer.ChannelBuffers; |
|||
import org.jboss.netty.channel.Channel; |
|||
import org.jboss.netty.channel.UpstreamMessageEvent; |
|||
import org.junit.Test; |
|||
|
|||
import com.corundumstudio.socketio.messages.PacketsMessage; |
|||
import com.corundumstudio.socketio.parser.Decoder; |
|||
import com.corundumstudio.socketio.parser.Encoder; |
|||
import com.corundumstudio.socketio.parser.Packet; |
|||
import com.corundumstudio.socketio.parser.PacketType; |
|||
|
|||
public class PacketHandlerTest { |
|||
|
|||
private ObjectMapper map = new ObjectMapper(); |
|||
private Decoder decoder = new Decoder(map); |
|||
private Encoder encoder = new Encoder(map); |
|||
@Mocked |
|||
private Channel channel; |
|||
|
|||
@Test |
|||
public void testOnePacket() throws Exception { |
|||
final AtomicInteger invocations = new AtomicInteger(); |
|||
PacketListener listener = new PacketListener(null, null, null) { |
|||
@Override |
|||
public void onPacket(Packet packet, SocketIOClient client) { |
|||
invocations.incrementAndGet(); |
|||
Assert.assertEquals(PacketType.JSON, packet.getType()); |
|||
Map<String, String> map = (Map<String, String>) packet.getData(); |
|||
Assert.assertTrue(map.keySet().size() == 1); |
|||
Assert.assertTrue(map.keySet().contains("test1")); |
|||
} |
|||
}; |
|||
PacketHandler handler = new PacketHandler(listener, decoder); |
|||
|
|||
List<Packet> packets = new ArrayList<Packet>(); |
|||
Packet packet = new Packet(PacketType.JSON); |
|||
packet.setData(Collections.singletonMap("test1", "test2")); |
|||
packets.add(packet); |
|||
|
|||
testHandler(invocations, handler, packets); |
|||
} |
|||
|
|||
@Test |
|||
public void testMultiplePackets() throws Exception { |
|||
final AtomicInteger invocations = new AtomicInteger(); |
|||
PacketListener listener = new PacketListener(null, null, null) { |
|||
@Override |
|||
public void onPacket(Packet packet, SocketIOClient client) { |
|||
if (packet.getType() == PacketType.CONNECT) { |
|||
invocations.incrementAndGet(); |
|||
return; |
|||
} |
|||
Assert.assertEquals(PacketType.JSON, packet.getType()); |
|||
Map<String, String> map = (Map<String, String>) packet.getData(); |
|||
Set<String> keys = new HashSet<String>(); |
|||
keys.add("test1"); |
|||
keys.add("fsdfdf"); |
|||
Assert.assertTrue(map.keySet().size() == 1); |
|||
Assert.assertTrue(map.keySet().removeAll(keys)); |
|||
invocations.incrementAndGet(); |
|||
} |
|||
}; |
|||
PacketHandler handler = new PacketHandler(listener, decoder); |
|||
|
|||
List<Packet> packets = new ArrayList<Packet>(); |
|||
Packet packet3 = new Packet(PacketType.CONNECT); |
|||
packets.add(packet3); |
|||
|
|||
Packet packet = new Packet(PacketType.JSON); |
|||
packet.setData(Collections.singletonMap("test1", "test2")); |
|||
packets.add(packet); |
|||
|
|||
Packet packet1 = new Packet(PacketType.JSON); |
|||
packet1.setData(Collections.singletonMap("fsdfdf", "wqeq")); |
|||
packets.add(packet1); |
|||
|
|||
testHandler(invocations, handler, packets); |
|||
} |
|||
|
|||
private void testHandler(final AtomicInteger invocations, |
|||
PacketHandler handler, List<Packet> packets) throws Exception { |
|||
String str = encoder.encodePackets(packets); |
|||
ChannelBuffer buffer = ChannelBuffers.wrappedBuffer(str.getBytes()); |
|||
handler.messageReceived(null, new UpstreamMessageEvent(channel, new PacketsMessage(null, buffer), null)); |
|||
Assert.assertEquals(packets.size(), invocations.get()); |
|||
} |
|||
|
|||
//@Test |
|||
public void testDecodePerf() throws Exception { |
|||
PacketListener listener = new PacketListener(null, null, null) { |
|||
@Override |
|||
public void onPacket(Packet packet, SocketIOClient client) { |
|||
} |
|||
}; |
|||
PacketHandler handler = new PacketHandler(listener, decoder); |
|||
long start = System.currentTimeMillis(); |
|||
ChannelBuffer buffer = ChannelBuffers.wrappedBuffer("\ufffd5\ufffd3:::5\ufffd7\ufffd3:::53d\ufffd3\ufffd0::\ufffd5\ufffd3:::5\ufffd7\ufffd3:::53d\ufffd3\ufffd0::\ufffd5\ufffd3:::5\ufffd7\ufffd3:::53d\ufffd3\ufffd0::\ufffd5\ufffd3:::5\ufffd7\ufffd3:::53d\ufffd3\ufffd0::\ufffd5\ufffd3:::5\ufffd7\ufffd3:::53d\ufffd3\ufffd0::".getBytes()); |
|||
for (int i = 0; i < 50000; i++) { |
|||
ChannelBuffer t = buffer.copy(); |
|||
handler.messageReceived(null, new UpstreamMessageEvent(channel, new PacketsMessage(null, t), null)); |
|||
} |
|||
long end = System.currentTimeMillis() - start; |
|||
System.out.println(end + "ms"); |
|||
// 1143ms |
|||
} |
|||
|
|||
|
|||
} |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue