X-Git-Url: https://git.pterodactylus.net/?a=blobdiff_plain;f=src%2Fnet%2Fpterodactylus%2Futil%2Ffcp%2FFcpConnection.java;h=9a132641619c364073401b246bbae6554f2d0dc0;hb=b905d1b51018f68ba5e8c598781a494921318c91;hp=846e135312de809439345bf87d9445af4935772d;hpb=0a4d1ff98574325df37f7cbb6b0655e86dd0cef4;p=jSite2.git diff --git a/src/net/pterodactylus/util/fcp/FcpConnection.java b/src/net/pterodactylus/util/fcp/FcpConnection.java index 846e135..9a13264 100644 --- a/src/net/pterodactylus/util/fcp/FcpConnection.java +++ b/src/net/pterodactylus/util/fcp/FcpConnection.java @@ -26,12 +26,13 @@ import java.net.InetAddress; import java.net.Socket; import java.net.UnknownHostException; import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; -import net.pterodactylus.util.fcp.message.CloseConnectionDuplicateClientName; -import net.pterodactylus.util.fcp.message.NodeHello; -import net.pterodactylus.util.fcp.message.SSKKeypair; import net.pterodactylus.util.io.Closer; +import net.pterodactylus.util.io.LimitedInputStream; /** * An FCP connection to a Freenet node. @@ -65,6 +66,20 @@ public class FcpConnection { /** The connection handler. */ private FcpConnectionHandler connectionHandler; + /** Incoming message statistics. */ + private Map incomingMessageStatistics = Collections.synchronizedMap(new HashMap()); + + /** + * Creates a new FCP connection to the freenet node running on localhost, + * using the default port. + * + * @throws UnknownHostException + * if the hostname can not be resolved + */ + public FcpConnection() throws UnknownHostException { + this(InetAddress.getLocalHost()); + } + /** * Creates a new FCP connection to the Freenet node running on the given * host, listening on the default port. @@ -184,6 +199,368 @@ public class FcpConnection { } /** + * Notifies listeners that a “Peer” message was received. + * + * @see FcpListener#receivedPeer(FcpConnection, Peer) + * @param peer + * The “Peer” message + */ + private void fireReceivedPeer(Peer peer) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedPeer(this, peer); + } + } + + /** + * Notifies all listeners that an “EndListPeers” message was received. + * + * @see FcpListener#receivedEndListPeers(FcpConnection, EndListPeers) + * @param endListPeers + * The “EndListPeers” message + */ + private void fireReceivedEndListPeers(EndListPeers endListPeers) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedEndListPeers(this, endListPeers); + } + } + + /** + * Notifies all listeners that a “PeerNote” message was received. + * + * @see FcpListener#receivedPeerNote(FcpConnection, PeerNote) + * @param peerNote + */ + private void fireReceivedPeerNote(PeerNote peerNote) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedPeerNote(this, peerNote); + } + } + + /** + * Notifies all listeners that an “EndListPeerNotes” message was received. + * + * @see FcpListener#receivedEndListPeerNotes(FcpConnection, + * EndListPeerNotes) + * @param endListPeerNotes + * The “EndListPeerNotes” message + */ + private void fireReceivedEndListPeerNotes(EndListPeerNotes endListPeerNotes) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedEndListPeerNotes(this, endListPeerNotes); + } + } + + /** + * Notifies all listeners that a “PeerRemoved” message was received. + * + * @see FcpListener#receivedPeerRemoved(FcpConnection, PeerRemoved) + * @param peerRemoved + * The “PeerRemoved” message + */ + private void fireReceivedPeerRemoved(PeerRemoved peerRemoved) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedPeerRemoved(this, peerRemoved); + } + } + + /** + * Notifies all listeners that a “NodeData” message was received. + * + * @see FcpListener#receivedNodeData(FcpConnection, NodeData) + * @param nodeData + * The “NodeData” message + */ + private void fireReceivedNodeData(NodeData nodeData) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedNodeData(this, nodeData); + } + } + + /** + * Notifies all listeners that a “TestDDAReply” message was received. + * + * @see FcpListener#receivedTestDDAReply(FcpConnection, TestDDAReply) + * @param testDDAReply + * The “TestDDAReply” message + */ + private void fireReceivedTestDDAReply(TestDDAReply testDDAReply) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedTestDDAReply(this, testDDAReply); + } + } + + /** + * Notifies all listeners that a “TestDDAComplete” message was received. + * + * @see FcpListener#receivedTestDDAComplete(FcpConnection, TestDDAComplete) + * @param testDDAComplete + * The “TestDDAComplete” message + */ + private void fireReceivedTestDDAComplete(TestDDAComplete testDDAComplete) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedTestDDAComplete(this, testDDAComplete); + } + } + + /** + * Notifies all listeners that a “PersistentGet” message was received. + * + * @param persistentGet + * The “PersistentGet” message + */ + private void fireReceivedPersistentGet(PersistentGet persistentGet) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedPersistentGet(this, persistentGet); + } + } + + /** + * Notifies all listeners that a “PersistentPut” message was received. + * + * @see FcpListener#receivedPersistentPut(FcpConnection, PersistentPut) + * @param persistentPut + * The “PersistentPut” message + */ + private void fireReceivedPersistentPut(PersistentPut persistentPut) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedPersistentPut(this, persistentPut); + } + } + + /** + * Notifies all listeners that a “EndListPersistentRequests” message was + * received. + * + * @param endListPersistentRequests + * The “EndListPersistentRequests” message + */ + private void fireReceivedEndListPersistentRequests(EndListPersistentRequests endListPersistentRequests) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedEndListPersistentRequests(this, endListPersistentRequests); + } + } + + /** + * Notifies all listeners that a “URIGenerated” message was received. + * + * @param uriGenerated + * The “URIGenerated” message + */ + private void fireReceivedURIGenerated(URIGenerated uriGenerated) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedURIGenerated(this, uriGenerated); + } + } + + /** + * Notifies all listeners that a “DataFound” message was received. + * + * @param dataFound + * The “DataFound” message + */ + private void fireReceivedDataFound(DataFound dataFound) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedDataFound(this, dataFound); + } + } + + /** + * Notifies all listeners that an “AllData” message was received. + * + * @param allData + * The “AllData” message + */ + private void fireReceivedAllData(AllData allData) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedAllData(this, allData); + } + } + + /** + * Notifies all listeners that a “SimpleProgress” message was received. + * + * @param simpleProgress + * The “SimpleProgress” message + */ + private void fireReceivedSimpleProgress(SimpleProgress simpleProgress) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedSimpleProgress(this, simpleProgress); + } + } + + /** + * Notifies all listeners that a “StartedCompression” message was received. + * + * @param startedCompression + * The “StartedCompression” message + */ + private void fireReceivedStartedCompression(StartedCompression startedCompression) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedStartedCompression(this, startedCompression); + } + } + + /** + * Notifies all listeners that a “FinishedCompression” message was received. + * + * @param finishedCompression + * The “FinishedCompression” message + */ + private void fireReceivedFinishedCompression(FinishedCompression finishedCompression) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receviedFinishedCompression(this, finishedCompression); + } + } + + /** + * Notifies all listeners that an “UnknownPeerNoteType” message was + * received. + * + * @param unknownPeerNoteType + * The “UnknownPeerNoteType” message + */ + private void fireReceivedUnknownPeerNoteType(UnknownPeerNoteType unknownPeerNoteType) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedUnknownPeerNoteType(this, unknownPeerNoteType); + } + } + + /** + * Notifies all listeners that an “UnknownNodeIdentifier” message was + * received. + * + * @param unknownNodeIdentifier + * The “UnknownNodeIdentifier” message + */ + private void fireReceivedUnknownNodeIdentifier(UnknownNodeIdentifier unknownNodeIdentifier) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedUnknownNodeIdentifier(this, unknownNodeIdentifier); + } + } + + /** + * Notifies all listeners that a “ConfigData” message was received. + * + * @param configData + * The “ConfigData” message + */ + private void fireReceivedConfigData(ConfigData configData) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedConfigData(this, configData); + } + } + + /** + * Notifies all listeners that a “GetFailed” message was received. + * + * @param getFailed + * The “GetFailed” message + */ + private void fireReceivedGetFailed(GetFailed getFailed) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedGetFailed(this, getFailed); + } + } + + /** + * Notifies all listeners that a “PutFailed” message was received. + * + * @param putFailed + * The “PutFailed” message + */ + private void fireReceivedPutFailed(PutFailed putFailed) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedPutFailed(this, putFailed); + } + } + + /** + * Notifies all listeners that an “IdentifierCollision” message was + * received. + * + * @param identifierCollision + * The “IdentifierCollision” message + */ + private void fireReceivedIdentifierCollision(IdentifierCollision identifierCollision) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedIdentifierCollision(this, identifierCollision); + } + } + + /** + * Notifies all listeners that an “PersistentPutDir” message was received. + * + * @param persistentPutDir + * The “PersistentPutDir” message + */ + private void fireReceivedPersistentPutDir(PersistentPutDir persistentPutDir) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedPersistentPutDir(this, persistentPutDir); + } + } + + /** + * Notifies all listeners that a “PersistentRequestRemoved” message was + * received. + * + * @param persistentRequestRemoved + * The “PersistentRequestRemoved” message + */ + private void fireReceivedPersistentRequestRemoved(PersistentRequestRemoved persistentRequestRemoved) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedPersistentRequestRemoved(this, persistentRequestRemoved); + } + } + + /** + * Notifies all listeners that a “SubscribedUSKUpdate” message was received. + * + * @param subscribedUSKUpdate + * The “SubscribedUSKUpdate” message + */ + private void fireReceivedSubscribedUSKUpdate(SubscribedUSKUpdate subscribedUSKUpdate) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedSubscribedUSKUpdate(this, subscribedUSKUpdate); + } + } + + /** + * Notifies all listeners that a “PluginInfo” message was received. + * + * @param pluginInfo + * The “PluginInfo” message + */ + private void fireReceivedPluginInfo(PluginInfo pluginInfo) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedPluginInfo(this, pluginInfo); + } + } + + /** + * Notifies all listeners that an “FCPPluginReply” message was received. + * + * @param fcpPluginReply + * The “FCPPluginReply” message + */ + private void fireReceivedFCPPluginReply(FCPPluginReply fcpPluginReply) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedFCPPluginReply(this, fcpPluginReply); + } + } + + /** + * Notifies all listeners that a “ProtocolError” message was received. + * + * @param protocolError + * The “ProtocolError” message + */ + private void fireReceivedProtocolError(ProtocolError protocolError) { + for (FcpListener fcpListener: fcpListeners) { + fcpListener.receivedProtocolError(this, protocolError); + } + } + + /** * Notifies all registered listeners that a message has been received. * * @see FcpListener#receivedMessage(FcpConnection, FcpMessage) @@ -257,8 +634,79 @@ public class FcpConnection { */ void handleMessage(FcpMessage fcpMessage) { String messageName = fcpMessage.getName(); - if ("SSKKeypair".equals(messageName)) { + countMessage(messageName); + if ("SimpleProgress".equals(messageName)) { + fireReceivedSimpleProgress(new SimpleProgress(fcpMessage)); + } else if ("ProtocolError".equals(messageName)) { + fireReceivedProtocolError(new ProtocolError(fcpMessage)); + } else if ("PersistentGet".equals(messageName)) { + fireReceivedPersistentGet(new PersistentGet(fcpMessage)); + } else if ("PersistentPut".equals(messageName)) { + fireReceivedPersistentPut(new PersistentPut(fcpMessage)); + } else if ("PersistentPutDir".equals(messageName)) { + fireReceivedPersistentPutDir(new PersistentPutDir(fcpMessage)); + } else if ("URIGenerated".equals(messageName)) { + fireReceivedURIGenerated(new URIGenerated(fcpMessage)); + } else if ("EndListPersistentRequests".equals(messageName)) { + fireReceivedEndListPersistentRequests(new EndListPersistentRequests(fcpMessage)); + } else if ("Peer".equals(messageName)) { + fireReceivedPeer(new Peer(fcpMessage)); + } else if ("PeerNote".equals(messageName)) { + fireReceivedPeerNote(new PeerNote(fcpMessage)); + } else if ("StartedCompression".equals(messageName)) { + fireReceivedStartedCompression(new StartedCompression(fcpMessage)); + } else if ("FinishedCompression".equals(messageName)) { + fireReceivedFinishedCompression(new FinishedCompression(fcpMessage)); + } else if ("GetFailed".equals(messageName)) { + fireReceivedGetFailed(new GetFailed(fcpMessage)); + } else if ("PutFailed".equals(messageName)) { + fireReceivedPutFailed(new PutFailed(fcpMessage)); + } else if ("DataFound".equals(messageName)) { + fireReceivedDataFound(new DataFound(fcpMessage)); + } else if ("SubscribedUSKUpdate".equals(messageName)) { + fireReceivedSubscribedUSKUpdate(new SubscribedUSKUpdate(fcpMessage)); + } else if ("IdentifierCollision".equals(messageName)) { + fireReceivedIdentifierCollision(new IdentifierCollision(fcpMessage)); + } else if ("AllData".equals(messageName)) { + LimitedInputStream payloadInputStream = getInputStream(FcpUtils.safeParseLong(fcpMessage.getField("DataLength"))); + fireReceivedAllData(new AllData(fcpMessage, payloadInputStream)); + try { + payloadInputStream.consume(); + } catch (IOException ioe1) { + /* well, ignore. when the connection handler fails, all fails. */ + } + } else if ("EndListPeerNotes".equals(messageName)) { + fireReceivedEndListPeerNotes(new EndListPeerNotes(fcpMessage)); + } else if ("EndListPeers".equals(messageName)) { + fireReceivedEndListPeers(new EndListPeers(fcpMessage)); + } else if ("SSKKeypair".equals(messageName)) { fireReceivedSSKKeypair(new SSKKeypair(fcpMessage)); + } else if ("PeerRemoved".equals(messageName)) { + fireReceivedPeerRemoved(new PeerRemoved(fcpMessage)); + } else if ("PersistentRequestRemoved".equals(messageName)) { + fireReceivedPersistentRequestRemoved(new PersistentRequestRemoved(fcpMessage)); + } else if ("UnknownPeerNoteType".equals(messageName)) { + fireReceivedUnknownPeerNoteType(new UnknownPeerNoteType(fcpMessage)); + } else if ("UnknownNodeIdentifier".equals(messageName)) { + fireReceivedUnknownNodeIdentifier(new UnknownNodeIdentifier(fcpMessage)); + } else if ("FCPPluginReply".equals(messageName)) { + LimitedInputStream payloadInputStream = getInputStream(FcpUtils.safeParseLong(fcpMessage.getField("DataLength"))); + fireReceivedFCPPluginReply(new FCPPluginReply(fcpMessage, payloadInputStream)); + try { + payloadInputStream.consume(); + } catch (IOException ioe1) { + /* ignore. */ + } + } else if ("PluginInfo".equals(messageName)) { + fireReceivedPluginInfo(new PluginInfo(fcpMessage)); + } else if ("NodeData".equals(messageName)) { + fireReceivedNodeData(new NodeData(fcpMessage)); + } else if ("TestDDAReply".equals(messageName)) { + fireReceivedTestDDAReply(new TestDDAReply(fcpMessage)); + } else if ("TestDDAComplete".equals(messageName)) { + fireReceivedTestDDAComplete(new TestDDAComplete(fcpMessage)); + } else if ("ConfigData".equals(messageName)) { + fireReceivedConfigData(new ConfigData(fcpMessage)); } else if ("NodeHello".equals(messageName)) { fireReceivedNodeHello(new NodeHello(fcpMessage)); } else if ("CloseConnectionDuplicateClientName".equals(messageName)) { @@ -268,4 +716,47 @@ public class FcpConnection { } } + /** + * Handles a disconnect from the node. + */ + synchronized void handleDisconnect() { + Closer.close(remoteInputStream); + Closer.close(remoteOutputStream); + Closer.close(remoteSocket); + connectionHandler = null; + } + + // + // PRIVATE METHODS + // + + /** + * Incremets the counter in {@link #incomingMessageStatistics} by 1 + * for the given message name. + * + * @param name + * The name of the message to count + */ + private void countMessage(String name) { + int oldValue = 0; + if (incomingMessageStatistics.containsKey(name)) { + oldValue = incomingMessageStatistics.get(name); + } + incomingMessageStatistics.put(name, oldValue + 1); + } + + /** + * Returns a limited input stream from the node’s input stream. + * + * @param dataLength + * The length of the stream + * @return The limited input stream + */ + private LimitedInputStream getInputStream(long dataLength) { + if (dataLength <= 0) { + return new LimitedInputStream(null, 0); + } + return new LimitedInputStream(remoteInputStream, dataLength); + } + }