+ *
+ *
+ *
+ * Instances of this class keep channel state content and must not be shared nor + * cached. + * + * @author Rogiel Josias Sulzbach + */ +public class PeerWireDecoder extends FrameDecoder { + private boolean handshaked = false; + + @Override + protected Object decode(ChannelHandlerContext ctx, Channel channel, + ChannelBuffer buffer) throws Exception { + buffer.markReaderIndex(); + + if (!handshaked) { + if (buffer.readableBytes() <= 47) // at least 47 bytes + return null; + + final int pstrlen = buffer.readByte(); + if (buffer.readableBytes() < pstrlen + 47) { + buffer.resetReaderIndex(); + return null; + } + buffer.readerIndex(buffer.readerIndex() - 1); + + final HandshakeMessage message = new HandshakeMessage(); + message.read(buffer); + handshaked = true; + + return message; + } else { + if (buffer.readableBytes() <= 4) { + buffer.resetReaderIndex(); + return null; + } + + int len = buffer.readInt(); + if (len == 0) { + return new KeepAliveMessage(); + } else if (buffer.readableBytes() < len) { + buffer.resetReaderIndex(); + return null; + } + + final byte id = buffer.readByte(); + final PeerWireReadableMessage message = getMessage(id); + if (message == null) + // force connection to be closed + throw new CorruptedFrameException("unknown message " + id); + message.read(buffer); + return message; + } + } + + /** + * Return the message represented by id. Will return null if + * message id is unknown. + * + * @param id + * the id of the message + * @return the message + */ + private PeerWireReadableMessage getMessage(byte id) { + PeerWireReadableMessage message = null; + switch (id) { + case BitfieldMessage.MESSAGE_ID: + message = new BitfieldMessage(); + break; + case CancelMessage.MESSAGE_ID: + message = new CancelMessage(); + break; + case ChokeMessage.MESSAGE_ID: + message = new ChokeMessage(); + break; + case HaveMessage.MESSAGE_ID: + message = new HaveMessage(); + break; + case InterestedMessage.MESSAGE_ID: + message = new InterestedMessage(); + break; + case NotInterestedMessage.MESSAGE_ID: + message = new NotInterestedMessage(); + break; + case PieceMessage.MESSAGE_ID: + message = new PieceMessage(); + break; + case PortMessage.MESSAGE_ID: + message = new PortMessage(); + break; + case RequestMessage.MESSAGE_ID: + message = new RequestMessage(); + break; + case UnchokeMessage.MESSAGE_ID: + message = new UnchokeMessage(); + break; + } + return message; + } +} diff --git a/src/main/java/net/torrent/protocol/peerwire/codec/PeerWireEncoder.java b/src/main/java/net/torrent/protocol/peerwire/codec/PeerWireEncoder.java new file mode 100644 index 0000000..d4534ad --- /dev/null +++ b/src/main/java/net/torrent/protocol/peerwire/codec/PeerWireEncoder.java @@ -0,0 +1,52 @@ +/* + * Copyright 2011 Rogiel Josias Sulzbach + * + * 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 net.torrent.protocol.peerwire.codec; + +import net.torrent.protocol.peerwire.message.HandshakeMessage; + +import org.jboss.netty.buffer.ChannelBuffer; +import org.jboss.netty.buffer.ChannelBuffers; +import org.jboss.netty.channel.Channel; +import org.jboss.netty.channel.ChannelHandlerContext; +import org.jboss.netty.handler.codec.oneone.OneToOneEncoder; + +/** + * Messages are encoded in {@link PeerWireWritableMessage#write(ChannelBuffer)} + * method. Message length is measured automatically by the encoder. + * + * @author Rogiel Josias Sulzbach + */ +public class PeerWireEncoder extends OneToOneEncoder { + @Override + protected Object encode(ChannelHandlerContext ctx, Channel channel, + Object msg) throws Exception { + ChannelBuffer buffer = ChannelBuffers.dynamicBuffer(); + if (!(msg instanceof PeerWireWritableMessage)) + return msg; + + if (msg instanceof HandshakeMessage) { + final HandshakeMessage message = (HandshakeMessage) msg; + message.write(buffer); + } else { + final PeerWireWritableMessage message = (PeerWireWritableMessage) msg; + buffer.writeInt(0); // allocate 4 bytes for header + message.write(buffer); + int len = buffer.readableBytes(); + buffer.setInt(0, len - 4); + } + return buffer; + } +} diff --git a/src/main/java/net/torrent/protocol/peerwire/handler/PeerWireCodecHandler.java b/src/main/java/net/torrent/protocol/peerwire/handler/PeerWireCodecHandler.java index 4390bea..079e05d 100644 --- a/src/main/java/net/torrent/protocol/peerwire/handler/PeerWireCodecHandler.java +++ b/src/main/java/net/torrent/protocol/peerwire/handler/PeerWireCodecHandler.java @@ -66,7 +66,8 @@ public class PeerWireCodecHandler extends SimpleChannelHandler { throws Exception { if (e.getMessage() instanceof HandshakeMessage) { if (state.hasHandshaked()) - throw new IllegalStateException("Handshake has already been sent"); + throw new IllegalStateException( + "Handshake has already been sent"); e.getFuture().addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) diff --git a/src/main/java/net/torrent/protocol/peerwire/handler/PeerWireManagerHeadHandler.java b/src/main/java/net/torrent/protocol/peerwire/handler/PeerWireManagerHeadHandler.java index ddad293..960c4e0 100644 --- a/src/main/java/net/torrent/protocol/peerwire/handler/PeerWireManagerHeadHandler.java +++ b/src/main/java/net/torrent/protocol/peerwire/handler/PeerWireManagerHeadHandler.java @@ -21,8 +21,8 @@ import net.torrent.protocol.peerwire.PeerWirePeer; import net.torrent.protocol.peerwire.manager.TorrentManager; import net.torrent.protocol.peerwire.message.HandshakeMessage; import net.torrent.torrent.context.TorrentPeer; -import net.torrent.torrent.context.TorrentPeerID; import net.torrent.torrent.context.TorrentPeerCapabilities.TorrentPeerCapability; +import net.torrent.torrent.context.TorrentPeerID; import org.jboss.netty.channel.Channel; import org.jboss.netty.channel.ChannelHandlerContext; diff --git a/src/main/java/net/torrent/protocol/peerwire/manager/PeerManager.java b/src/main/java/net/torrent/protocol/peerwire/manager/PeerManager.java index 4638ad0..a1c3e8a 100644 --- a/src/main/java/net/torrent/protocol/peerwire/manager/PeerManager.java +++ b/src/main/java/net/torrent/protocol/peerwire/manager/PeerManager.java @@ -25,6 +25,7 @@ import java.util.Set; import net.torrent.protocol.peerwire.PeerWirePeer; import net.torrent.torrent.context.TorrentContext; import net.torrent.torrent.context.TorrentPeer; +import net.torrent.util.PeerCallback; import net.torrent.util.PeerWirePeerCallback; import org.jboss.netty.channel.Channel; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/CancelMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/CancelMessage.java index 14dcab0..27fa38b 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/CancelMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/CancelMessage.java @@ -74,7 +74,7 @@ public class CancelMessage implements PeerWireWritableMessage, buffer.writeInt(start); buffer.writeInt(length); } - + @Override public int length() { return 3 * 4 + 1; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/ChokeMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/ChokeMessage.java index a1a02f9..27b2a4f 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/ChokeMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/ChokeMessage.java @@ -44,7 +44,7 @@ public class ChokeMessage implements PeerWireWritableMessage, public void write(ChannelBuffer buffer) throws IOException { buffer.writeByte(MESSAGE_ID); } - + @Override public int length() { return 1; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/HandshakeMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/HandshakeMessage.java index b088e8c..e4bfb40 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/HandshakeMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/HandshakeMessage.java @@ -101,7 +101,8 @@ public class HandshakeMessage implements PeerWireWritableMessage, @Override public int length() { - return 1 + pstrlen + pstr.length() + 8 + infohash.length + peerId.length; + return 1 + pstrlen + pstr.length() + 8 + infohash.length + + peerId.length; } public int getPstrlen() { diff --git a/src/main/java/net/torrent/protocol/peerwire/message/HaveMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/HaveMessage.java index 41ac4ce..573dcbc 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/HaveMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/HaveMessage.java @@ -71,7 +71,7 @@ public class HaveMessage implements PeerWireWritableMessage, buffer.writeByte(MESSAGE_ID); buffer.writeInt(piece); } - + @Override public int length() { return 5; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/InterestedMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/InterestedMessage.java index a70442f..1f42a59 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/InterestedMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/InterestedMessage.java @@ -44,7 +44,7 @@ public class InterestedMessage implements PeerWireWritableMessage, public void write(ChannelBuffer buffer) throws IOException { buffer.writeByte(MESSAGE_ID); } - + @Override public int length() { return 1; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/KeepAliveMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/KeepAliveMessage.java index 6d63356..0d09c6d 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/KeepAliveMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/KeepAliveMessage.java @@ -46,7 +46,7 @@ public class KeepAliveMessage implements PeerWireWritableMessage, @Override public void write(ChannelBuffer buffer) throws IOException { } - + @Override public int length() { return 0; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/NotInterestedMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/NotInterestedMessage.java index 240474c..f188860 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/NotInterestedMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/NotInterestedMessage.java @@ -44,7 +44,7 @@ public class NotInterestedMessage implements PeerWireWritableMessage, public void write(ChannelBuffer buffer) throws IOException { buffer.writeByte(MESSAGE_ID); } - + @Override public int length() { return 1; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/PieceMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/PieceMessage.java index 859894e..d518a79 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/PieceMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/PieceMessage.java @@ -90,7 +90,7 @@ public class PieceMessage implements PeerWireWritableMessage, buffer.writeInt(start); buffer.writeBytes(block); } - + @Override public int length() { return 9 + block.capacity(); diff --git a/src/main/java/net/torrent/protocol/peerwire/message/PortMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/PortMessage.java index ac7f200..c6911b2 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/PortMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/PortMessage.java @@ -61,7 +61,7 @@ public class PortMessage implements PeerWireWritableMessage, buffer.writeByte(MESSAGE_ID); buffer.writeShort(port); } - + @Override public int length() { return 3; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/RequestMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/RequestMessage.java index b4329eb..4c1a531 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/RequestMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/RequestMessage.java @@ -112,7 +112,7 @@ public class RequestMessage implements PeerWireWritableMessage, buffer.writeInt(start); buffer.writeInt(length); } - + @Override public int length() { return 13; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/UnchokeMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/UnchokeMessage.java index 97d04d8..44edbe8 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/UnchokeMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/UnchokeMessage.java @@ -44,7 +44,7 @@ public class UnchokeMessage implements PeerWireWritableMessage, public void write(ChannelBuffer buffer) throws IOException { buffer.writeByte(MESSAGE_ID); } - + @Override public int length() { return 1; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/fast/AllowedFastMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/fast/AllowedFastMessage.java index 831dc7b..e6107a8 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/fast/AllowedFastMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/fast/AllowedFastMessage.java @@ -97,7 +97,7 @@ public class AllowedFastMessage implements PeerWireWritableMessage, buffer.writeInt(piece); } } - + @Override public int length() { return 1 + pieces.length * 4; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/fast/HaveAllMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/fast/HaveAllMessage.java index b9985d7..3438ea3 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/fast/HaveAllMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/fast/HaveAllMessage.java @@ -52,7 +52,7 @@ public class HaveAllMessage implements PeerWireWritableMessage, public void write(ChannelBuffer buffer) throws IOException { buffer.writeByte(MESSAGE_ID); } - + @Override public int length() { return 1; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/fast/HaveNoneMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/fast/HaveNoneMessage.java index 289f2eb..43cae2a 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/fast/HaveNoneMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/fast/HaveNoneMessage.java @@ -52,7 +52,7 @@ public class HaveNoneMessage implements PeerWireWritableMessage, public void write(ChannelBuffer buffer) throws IOException { buffer.writeByte(MESSAGE_ID); } - + @Override public int length() { return 1; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/fast/RejectMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/fast/RejectMessage.java index 62ae8be..8b8a64c 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/fast/RejectMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/fast/RejectMessage.java @@ -94,7 +94,7 @@ public class RejectMessage implements PeerWireWritableMessage, buffer.writeInt(start); buffer.writeInt(length); } - + @Override public int length() { return 13; diff --git a/src/main/java/net/torrent/protocol/peerwire/message/fast/SuggestPieceMessage.java b/src/main/java/net/torrent/protocol/peerwire/message/fast/SuggestPieceMessage.java index 7077a7b..5dfd1c5 100644 --- a/src/main/java/net/torrent/protocol/peerwire/message/fast/SuggestPieceMessage.java +++ b/src/main/java/net/torrent/protocol/peerwire/message/fast/SuggestPieceMessage.java @@ -69,7 +69,7 @@ public class SuggestPieceMessage implements PeerWireWritableMessage, buffer.writeByte(MESSAGE_ID); buffer.writeInt(piece); } - + @Override public int length() { return 5; diff --git a/src/main/java/net/torrent/torrent/context/TorrentContext.java b/src/main/java/net/torrent/torrent/context/TorrentContext.java index 76f7c89..cb4258a 100644 --- a/src/main/java/net/torrent/torrent/context/TorrentContext.java +++ b/src/main/java/net/torrent/torrent/context/TorrentContext.java @@ -20,17 +20,19 @@ import java.util.Collections; import java.util.HashSet; import java.util.Set; -import net.torrent.protocol.tracker.message.PeerListMessage.PeerInfo; import net.torrent.torrent.Torrent; import net.torrent.torrent.context.TorrentPeerCapabilities.TorrentPeerCapability; +/** + * @author Rogiel Josias Sulzbach + */ public class TorrentContext { /** * The torrent metadata object */ private final Torrent torrent; /** - * The bitfield + * The current torrent bitfield */ private final TorrentBitfield bitfield = new TorrentBitfield(this); @@ -116,14 +118,38 @@ public class TorrentContext { return Collections.unmodifiableSet(unknownPeers); } + /** + * Get the peer with the given id + * + * @param peerId + * the peer id + * @return the found peer + */ public TorrentPeer getPeer(TorrentPeerID peerId) { return swarm.getPeer(peerId); } + /** + * Get the peer with the given address + * + * @param address + * the address + * @return the found peer + */ public TorrentPeer getPeer(InetSocketAddress address) { return swarm.getPeer(address); } + /** + * Try to locate a peer first by its id. If no match is found, tries to + * lookup by its address. + * + * @param id + * the peer id + * @param address + * the peer address + * @return the found peer + */ public TorrentPeer getPeer(TorrentPeerID id, InetSocketAddress address) { return swarm.getPeer(id, address); } diff --git a/src/main/java/net/torrent/torrent/context/TorrentSwarm.java b/src/main/java/net/torrent/torrent/context/TorrentSwarm.java index af915a4..476b09b 100644 --- a/src/main/java/net/torrent/torrent/context/TorrentSwarm.java +++ b/src/main/java/net/torrent/torrent/context/TorrentSwarm.java @@ -1,5 +1,17 @@ -/** - * +/* + * Copyright 2011 Rogiel Josias Sulzbach + * + * 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 net.torrent.torrent.context;