r27005 - trunk/freenet/src/freenet/node/fcp

Author: toad
Date: 2009-04-18 18:29:30 +0000 (Sat, 18 Apr 2009)
New Revision: 27005

Added:
   trunk/freenet/src/freenet/node/fcp/UnsubscribeUSKMessage.java
Modified:
   trunk/freenet/src/freenet/node/fcp/FCPConnectionHandler.java
   trunk/freenet/src/freenet/node/fcp/FCPMessage.java
   trunk/freenet/src/freenet/node/fcp/SubscribeUSK.java
   trunk/freenet/src/freenet/node/fcp/SubscribeUSKMessage.java
Log:
Track subscribed USKs by identifier, add UnsubscribeUSK, also unsubscribe on disconnect (fix leak)


Modified: trunk/freenet/src/freenet/node/fcp/FCPConnectionHandler.java
===================================================================
--- trunk/freenet/src/freenet/node/fcp/FCPConnectionHandler.java	2009-04-18 18:10:07 UTC (rev 27004)
+++ trunk/freenet/src/freenet/node/fcp/FCPConnectionHandler.java	2009-04-18 18:29:30 UTC (rev 27005)
@@ -9,6 +9,7 @@
 import java.net.Socket;
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.Map;
 import java.util.Random;
 
 import com.db4o.ObjectContainer;
@@ -59,6 +60,7 @@
 	final FCPServer server;
 	final Socket sock;
 	final FCPConnectionInputHandler inputHandler;
+	final Map<String, SubscribeUSK> uskSubscriptions;
 	public final FCPConnectionOutputHandler outputHandler;
 	private boolean isClosed;
 	private boolean inputClosed;
@@ -92,6 +94,7 @@
 		isClosed = false;
 		this.bf = server.core.tempBucketFactory;
 		requestsByIdentifier = new HashMap<String, ClientRequest>();
+		uskSubscriptions = new HashMap<String, SubscribeUSK>();
 		this.inputHandler = new FCPConnectionInputHandler(this);
 		this.outputHandler = new FCPConnectionOutputHandler(this);
 		
@@ -112,14 +115,18 @@
 		if(foreverClient != null)
 			foreverClient.onLostConnection(this);
 		boolean dupe;
+		SubscribeUSK[] subscriptions;
 		synchronized(this) {
 			isClosed = true;
 			requests = new ClientRequest[requestsByIdentifier.size()];
 			requests = requestsByIdentifier.values().toArray(requests);
+			subscriptions = uskSubscriptions.values().toArray(new SubscribeUSK[uskSubscriptions.size()]);
 			dupe = killedDupe;
 		}
 		for(int i=0;i<requests.length;i++)
 			requests[i].onLostConnection(null, server.core.clientContext);
+		for(SubscribeUSK sub : subscriptions)
+			sub.unsubscribe();
 		if(!dupe) {
 		server.core.clientContext.jobRunner.queue(new DBJob() {
 
@@ -140,6 +147,7 @@
 			
 		}, NativeThread.NORM_PRIORITY, false);
 		}
+		
 		outputHandler.onClosed();
 	}
 	
@@ -738,6 +746,19 @@
 		Logger.error(this, "Not storing FCPConnectionHandler in database", new Exception("error"));
 		return false;
 	}
-	
 
+	public synchronized void addSubscription(String identifier, SubscribeUSK subscribeUSK) throws IdentifierCollisionException {
+		if(uskSubscriptions.containsKey(identifier)) throw new IdentifierCollisionException();
+		uskSubscriptions.put(identifier, subscribeUSK);
+	}
+
+	public void unsubscribe(String identifier) throws MessageInvalidException {
+		SubscribeUSK sub;
+		synchronized(this) {
+			if(!uskSubscriptions.containsKey(identifier)) throw new MessageInvalidException(ProtocolErrorMessage.NO_SUCH_IDENTIFIER, "No such identifier unsubscribing", identifier, false);
+			sub = uskSubscriptions.remove(identifier);
+		}
+		sub.unsubscribe();
+	}
+
 }

Modified: trunk/freenet/src/freenet/node/fcp/FCPMessage.java
===================================================================
--- trunk/freenet/src/freenet/node/fcp/FCPMessage.java	2009-04-18 18:10:07 UTC (rev 27004)
+++ trunk/freenet/src/freenet/node/fcp/FCPMessage.java	2009-04-18 18:29:30 UTC (rev 27005)
@@ -86,6 +86,8 @@
 			return new ShutdownMessage();
 		if(name.equals(SubscribeUSKMessage.NAME))
 			return new SubscribeUSKMessage(fs);
+		if(name.equals(UnsubscribeUSKMessage.NAME))
+			return new UnsubscribeUSKMessage(fs);
 		if(name.equals(TestDDARequestMessage.NAME))
 			return new TestDDARequestMessage(fs);
 		if(name.equals(TestDDAResponseMessage.NAME))

Modified: trunk/freenet/src/freenet/node/fcp/SubscribeUSK.java
===================================================================
--- trunk/freenet/src/freenet/node/fcp/SubscribeUSK.java	2009-04-18 18:10:07 UTC (rev 27004)
+++ trunk/freenet/src/freenet/node/fcp/SubscribeUSK.java	2009-04-18 18:29:30 UTC (rev 27005)
@@ -19,14 +19,17 @@
 	final boolean dontPoll;
 	final short prio;
 	final short prioProgress;
+	final USK usk;
 	
-	public SubscribeUSK(SubscribeUSKMessage message, NodeClientCore core, FCPConnectionHandler handler) {
+	public SubscribeUSK(SubscribeUSKMessage message, NodeClientCore core, FCPConnectionHandler handler) throws IdentifierCollisionException {
 		this.handler = handler;
 		this.dontPoll = message.dontPoll;
 		this.identifier = message.identifier;
 		this.core = core;
+		this.usk = message.key;
 		prio = message.prio;
 		prioProgress = message.prioProgress;
+		handler.addSubscription(identifier, this);
 		core.uskManager.subscribe(message.key, this, !message.dontPoll, handler.getRebootClient().lowLevelClient);
 	}
 
@@ -48,4 +51,8 @@
 		return prioProgress;
 	}
 
+	public void unsubscribe() {
+		core.uskManager.unsubscribe(usk, this);
+	}
+
 }

Modified: trunk/freenet/src/freenet/node/fcp/SubscribeUSKMessage.java
===================================================================
--- trunk/freenet/src/freenet/node/fcp/SubscribeUSKMessage.java	2009-04-18 18:10:07 UTC (rev 27004)
+++ trunk/freenet/src/freenet/node/fcp/SubscribeUSKMessage.java	2009-04-18 18:29:30 UTC (rev 27005)
@@ -70,8 +70,12 @@
 	@Override
 	public void run(FCPConnectionHandler handler, Node node)
 			throws MessageInvalidException {
-		new SubscribeUSK(this, node.clientCore, handler);
-		
+		try {
+			new SubscribeUSK(this, node.clientCore, handler);
+		} catch (IdentifierCollisionException e) {
+			handler.outputHandler.queue(new IdentifierCollisionMessage(identifier, false));
+			return;
+		}
 		SubscribedUSKMessage reply = new SubscribedUSKMessage(this);
 		handler.outputHandler.queue(reply);
 	}

Added: trunk/freenet/src/freenet/node/fcp/UnsubscribeUSKMessage.java
===================================================================
--- trunk/freenet/src/freenet/node/fcp/UnsubscribeUSKMessage.java	                        (rev 0)
+++ trunk/freenet/src/freenet/node/fcp/UnsubscribeUSKMessage.java	2009-04-18 18:29:30 UTC (rev 27005)
@@ -0,0 +1,40 @@
+package freenet.node.fcp;
+
+import com.db4o.ObjectContainer;
+
+import freenet.node.Node;
+import freenet.support.SimpleFieldSet;
+
+public class UnsubscribeUSKMessage extends FCPMessage {
+
+	public static final String NAME = "UnsubscribeUSK";
+	private final String identifier;
+
+	public UnsubscribeUSKMessage(SimpleFieldSet fs) throws MessageInvalidException {
+		this.identifier = fs.get("Identifier");
+		if(identifier == null)
+			throw new MessageInvalidException(ProtocolErrorMessage.MISSING_FIELD, "No Identifier!", null, false);
+	}
+
+	@Override
+	public SimpleFieldSet getFieldSet() {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public String getName() {
+		return NAME;
+	}
+
+	@Override
+	public void removeFrom(ObjectContainer container) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public void run(FCPConnectionHandler handler, Node node)
+			throws MessageInvalidException {
+		handler.unsubscribe(identifier);
+	}
+
+}