[e-cvs] cvs commit: e/src/jsrc/net/vattp/data VatTPConnection.java VatTPMgr.java AuthSecrets.java DataCommThunk.java DataPath.java ListenThread.java Msg.java MsgHandler.java NetConfig.java NewConnectionReactor.java RecvThread.java StartUpProtocol.java Suspend.java VatLocationLookup.java ConnectionsMgr.java DataConnection.java
markm@eros.cs.jhu.edu
markm@eros.cs.jhu.edu
Mon, 1 Oct 2001 21:29:18 -0400
markm 01/10/01 21:29:18
Modified: src/jsrc/net/captp/jcomm CapTPReplacer.java
CapTPResolver.java DelayedRedirector.java
Far3Desc.java FarHandler.java ImportDesc.java
IncomingDesc.java Introducer.java LocatorUnum.java
LookupHandler.java NewFarDesc.java
NewRemotePromiseDesc.java NonceLocator.java
ObjectRefDesc.java Promise3Desc.java
RemoteHandler.java RemotePromiseHandler.java
src/jsrc/net/vattp/data AuthSecrets.java DataCommThunk.java
DataPath.java ListenThread.java Msg.java
MsgHandler.java NetConfig.java
NewConnectionReactor.java RecvThread.java
StartUpProtocol.java Suspend.java
VatLocationLookup.java
Added: src/jsrc/net/captp/jcomm CapTPConnection.java CapTPMgr.java
src/jsrc/net/vattp/data VatTPConnection.java VatTPMgr.java
Removed: src/jsrc/net/captp/jcomm ProxyConnection.java ProxyMgr.java
src/jsrc/net/vattp/data ConnectionsMgr.java
DataConnection.java
Log:
various renamings
Revision Changes Path
1.4 +2 -2 e/src/jsrc/net/captp/jcomm/CapTPReplacer.java
Index: CapTPReplacer.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/CapTPReplacer.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- CapTPReplacer.java 2001/09/06 09:55:38 1.3
+++ CapTPReplacer.java 2001/10/02 01:29:17 1.4
@@ -40,12 +40,12 @@
/**
* The connection over which we are communicating
*/
- private ProxyConnection myConn;
+ private CapTPConnection myConn;
/**
*
*/
- /*package*/ CapTPReplacer(ProxyConnection conn) {
+ /*package*/ CapTPReplacer(CapTPConnection conn) {
myConn = conn;
}
1.4 +3 -3 e/src/jsrc/net/captp/jcomm/CapTPResolver.java
Index: CapTPResolver.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/CapTPResolver.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- CapTPResolver.java 2001/09/06 09:55:38 1.3
+++ CapTPResolver.java 2001/10/02 01:29:17 1.4
@@ -35,19 +35,19 @@
/**
* The connection over which we are communicating
*/
- private ProxyConnection myConn;
+ private CapTPConnection myConn;
/**
*
*/
- /*package*/ CapTPResolver(ProxyConnection conn) {
+ /*package*/ CapTPResolver(CapTPConnection conn) {
myConn = conn;
}
/**
* Replace ObjectRefDescs in the input stream with the object reference
- * they {@link ObjectRefDesc#dereference(ProxyConnection)} to.
+ * they {@link ObjectRefDesc#dereference(CapTPConnection)} to.
*/
public Object run(Object ref) {
ref = Ref.resolution(ref);
1.8 +1 -1 e/src/jsrc/net/captp/jcomm/DelayedRedirector.java
Index: DelayedRedirector.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/DelayedRedirector.java,v
retrieving revision 1.7
retrieving revision 1.8
diff -u -r1.7 -r1.8
--- DelayedRedirector.java 2001/09/06 09:55:38 1.7
+++ DelayedRedirector.java 2001/10/02 01:29:17 1.8
@@ -84,7 +84,7 @@
return null;
}
- ProxyConnection conn = rvh.myConn;
+ CapTPConnection conn = rvh.myConn;
Unsealer unsealer = conn.getUnsealer();
Brand brand = unsealer.brand();
SealedBox optTargetBox = Ref.optMeta(target, brand);
1.6 +1 -1 e/src/jsrc/net/captp/jcomm/Far3Desc.java
Index: Far3Desc.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/Far3Desc.java,v
retrieving revision 1.5
retrieving revision 1.6
diff -u -r1.5 -r1.6
--- Far3Desc.java 2001/09/06 09:55:38 1.5
+++ Far3Desc.java 2001/10/02 01:29:17 1.6
@@ -66,7 +66,7 @@
* What the other side imported (from somebody other than ourselves), we
* dereference as a FarRef to the same object.
*/
- public Object dereference(ProxyConnection conn) {
+ public Object dereference(CapTPConnection conn) {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(conn + " deref Far3Desc: " + this);
}
1.5 +2 -2 e/src/jsrc/net/captp/jcomm/FarHandler.java
Index: FarHandler.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/FarHandler.java,v
retrieving revision 1.4
retrieving revision 1.5
diff -u -r1.4 -r1.5
--- FarHandler.java 2001/09/06 09:55:38 1.4
+++ FarHandler.java 2001/10/02 01:29:17 1.5
@@ -34,12 +34,12 @@
/**
* Construct a new FarHandler.
*
- * @param connection The ProxyConnection to communicate via
+ * @param connection The CapTPConnection to communicate via
* @param pos The Imports or Questions map pos of the object
* @param swissHash The identity (within the connection's remote vat) of
* the object this FarRef designates.
*/
- /*package*/ FarHandler(ProxyConnection conn,
+ /*package*/ FarHandler(CapTPConnection conn,
int pos,
BigInteger swissHash)
{
1.6 +1 -1 e/src/jsrc/net/captp/jcomm/ImportDesc.java
Index: ImportDesc.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/ImportDesc.java,v
retrieving revision 1.5
retrieving revision 1.6
diff -u -r1.5 -r1.6
--- ImportDesc.java 2001/09/06 09:55:38 1.5
+++ ImportDesc.java 2001/10/02 01:29:17 1.6
@@ -57,7 +57,7 @@
/**
* What the other side exported, we dereference as what we imported.
*/
- public Object dereference(ProxyConnection conn) {
+ public Object dereference(CapTPConnection conn) {
validate();
Ref result = conn.getImport(myImportPos);
1.6 +2 -2 e/src/jsrc/net/captp/jcomm/IncomingDesc.java
Index: IncomingDesc.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/IncomingDesc.java,v
retrieving revision 1.5
retrieving revision 1.6
diff -u -r1.5 -r1.6
--- IncomingDesc.java 2001/09/06 09:55:38 1.5
+++ IncomingDesc.java 2001/10/02 01:29:17 1.6
@@ -59,9 +59,9 @@
/**
* What the other side imported, we dereference as an object we exported.
*/
- public Object dereference(ProxyConnection conn) {
+ public Object dereference(CapTPConnection conn) {
if (Trace.captp.debug && Trace.ON) {
- Trace.captp.debugm("ProxyConnection " + conn +
+ Trace.captp.debugm("CapTPConnection " + conn +
" deref IncomingDesc, pos=" + myIncomingPos);
}
return conn.getIncoming(myIncomingPos);
1.14 +1 -1 e/src/jsrc/net/captp/jcomm/Introducer.java
Index: Introducer.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/Introducer.java,v
retrieving revision 1.13
retrieving revision 1.14
diff -u -r1.13 -r1.14
--- Introducer.java 2001/09/07 05:49:20 1.13
+++ Introducer.java 2001/10/02 01:29:17 1.14
@@ -211,7 +211,7 @@
}
/* Turn on the comm system... */
- ProxyMgr proxyMgr = new ProxyMgr(myOptVatIdentity,
+ CapTPMgr proxyMgr = new CapTPMgr(myOptVatIdentity,
myNetConfig,
myOptSwissTable,
myEntropy);
1.11 +3 -3 e/src/jsrc/net/captp/jcomm/LocatorUnum.java
Index: LocatorUnum.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/LocatorUnum.java,v
retrieving revision 1.10
retrieving revision 1.11
diff -u -r1.10 -r1.11
--- LocatorUnum.java 2001/09/23 05:14:48 1.10
+++ LocatorUnum.java 2001/10/02 01:29:17 1.11
@@ -65,12 +65,12 @@
/**
* XXX Need to recover a valid value somehow on revival
*/
- private transient ProxyMgr myProxyMgr;
+ private transient CapTPMgr myProxyMgr;
/**
*
*/
- /*package*/ LocatorUnum(SwissTable swissTable, ProxyMgr proxyMgr) {
+ /*package*/ LocatorUnum(SwissTable swissTable, CapTPMgr proxyMgr) {
mySwissTable = swissTable;
myProxyMgr = proxyMgr;
}
@@ -98,7 +98,7 @@
("" + this + ".getRcvr(" + searchPath + ", " + vatID +
", " + swissNum + ", " + optFarVine + ")");
}
- ProxyConnection optProxyConn =
+ CapTPConnection optProxyConn =
myProxyMgr.getOrMakeProxyConnection(searchPath, vatID);
if (optProxyConn == null) {
1.4 +5 -5 e/src/jsrc/net/captp/jcomm/LookupHandler.java
Index: LookupHandler.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/LookupHandler.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- LookupHandler.java 2001/09/06 09:55:38 1.3
+++ LookupHandler.java 2001/10/02 01:29:17 1.4
@@ -29,11 +29,11 @@
/*package*/ class LookupHandler extends RemoteHandler {
/**
- * Construct a new LookupHandler.
- *
- * @param connection The ProxyConnection to communicate via.
- */
- /*package*/ LookupHandler(ProxyConnection conn) {
+ * Construct a new LookupHandler.
+ *
+ * @param connection The CapTPConnection to communicate via.
+ */
+ /*package*/ LookupHandler(CapTPConnection conn) {
super(conn, 0, null);
}
}
1.7 +1 -1 e/src/jsrc/net/captp/jcomm/NewFarDesc.java
Index: NewFarDesc.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/NewFarDesc.java,v
retrieving revision 1.6
retrieving revision 1.7
diff -u -r1.6 -r1.7
--- NewFarDesc.java 2001/09/06 09:55:38 1.6
+++ NewFarDesc.java 2001/10/02 01:29:17 1.7
@@ -69,7 +69,7 @@
* What the other side exported, we dereference as the "new" Far
* reference we will now import.
*/
- public Object dereference(ProxyConnection conn) {
+ public Object dereference(CapTPConnection conn) {
validate();
Ref result = conn.newFarRef(myImportPos, mySwissHash);
1.4 +2 -2 e/src/jsrc/net/captp/jcomm/NewRemotePromiseDesc.java
Index: NewRemotePromiseDesc.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/NewRemotePromiseDesc.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- NewRemotePromiseDesc.java 2001/09/06 09:55:38 1.3
+++ NewRemotePromiseDesc.java 2001/10/02 01:29:17 1.4
@@ -74,14 +74,14 @@
* What the other side exported, we dereference as the RemotePromise
* we will now import.
*/
- public Object dereference(ProxyConnection conn) {
+ public Object dereference(CapTPConnection conn) {
Object result = conn.newRemotePromise(myImportPos,
myRdrPos,
myRdrBase);
if (Trace.captp.debug && Trace.ON) {
- Trace.captp.debugm("ProxyConnection " + conn +
+ Trace.captp.debugm("CapTPConnection " + conn +
" deref " + toString());
}
return result;
1.9 +4 -4 e/src/jsrc/net/captp/jcomm/NonceLocator.java
Index: NonceLocator.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/NonceLocator.java,v
retrieving revision 1.8
retrieving revision 1.9
diff -u -r1.8 -r1.9
--- NonceLocator.java 2001/10/01 08:31:04 1.8
+++ NonceLocator.java 2001/10/02 01:29:17 1.9
@@ -61,7 +61,7 @@
/**
*
*/
- private ProxyMgr myProxyMgr;
+ private CapTPMgr myProxyMgr;
/**
*
@@ -74,7 +74,7 @@
/*package*/ NonceLocator(PromiseGiftTable pGifts,
NearGiftTable nGifts,
String ownID,
- ProxyMgr proxyMgr,
+ CapTPMgr proxyMgr,
SwissTable swissTable)
{
myPGifts = pGifts;
@@ -128,7 +128,7 @@
BigInteger nonce,
Object optFarVine)
throws IOException {
- ProxyConnection optDonorConn = myProxyMgr.optProxyConnection(donorID);
+ CapTPConnection optDonorConn = myProxyMgr.optProxyConnection(donorID);
if (null == optDonorConn) {
return Ref.broken(E.asRTE("The donor is gone"));
}
@@ -153,7 +153,7 @@
BigInteger swissHash,
Object optFarVine)
throws IOException {
- ProxyConnection optDonorConn = myProxyMgr.optProxyConnection(donorID);
+ CapTPConnection optDonorConn = myProxyMgr.optProxyConnection(donorID);
if (null == optDonorConn) {
return Ref.broken(E.asRTE("The donor is gone"));
}
1.7 +1 -1 e/src/jsrc/net/captp/jcomm/ObjectRefDesc.java
Index: ObjectRefDesc.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/ObjectRefDesc.java,v
retrieving revision 1.6
retrieving revision 1.7
diff -u -r1.6 -r1.7
--- ObjectRefDesc.java 2001/09/06 09:55:38 1.6
+++ ObjectRefDesc.java 2001/10/02 01:29:17 1.7
@@ -44,5 +44,5 @@
*
* @param conn The connection over which we are communicating
*/
- /*package*/ Object dereference(ProxyConnection conn);
+ /*package*/ Object dereference(CapTPConnection conn);
}
1.4 +1 -1 e/src/jsrc/net/captp/jcomm/Promise3Desc.java
Index: Promise3Desc.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/Promise3Desc.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- Promise3Desc.java 2001/09/06 09:55:38 1.3
+++ Promise3Desc.java 2001/10/02 01:29:17 1.4
@@ -62,7 +62,7 @@
* ourselves), we dereference as a RemotePromise to the same
* arrowhead.
*/
- public Object dereference(ProxyConnection conn) {
+ public Object dereference(CapTPConnection conn) {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(conn + " deref Promise3Desc: " + this);
}
1.16 +7 -7 e/src/jsrc/net/captp/jcomm/RemoteHandler.java
Index: RemoteHandler.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/RemoteHandler.java,v
retrieving revision 1.15
retrieving revision 1.16
diff -u -r1.15 -r1.16
--- RemoteHandler.java 2001/09/15 02:03:41 1.15
+++ RemoteHandler.java 2001/10/02 01:29:17 1.16
@@ -53,7 +53,7 @@
/**
* Connection to the vat I point into
*/
- /*package*/ ProxyConnection myConn;
+ /*package*/ CapTPConnection myConn;
/**
* My outgoing position. <p>
@@ -82,12 +82,12 @@
/*package*/ FlexList myBreakageReactors;
/**
- * @param connection The ProxyConnection to communicate via
- * @param pos The Imports or Questions map pos of the object
- * @param optIdentity null, or the sameness identity of the far
- * reference.
- */
- /*package*/ RemoteHandler(ProxyConnection connection,
+ * @param connection The CapTPConnection to communicate via
+ * @param pos The Imports or Questions map pos of the object
+ * @param optIdentity null, or the sameness identity of the far
+ * reference.
+ */
+ /*package*/ RemoteHandler(CapTPConnection connection,
int pos,
Object optIdentity)
{
1.3 +6 -6 e/src/jsrc/net/captp/jcomm/RemotePromiseHandler.java
Index: RemotePromiseHandler.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/captp/jcomm/RemotePromiseHandler.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- RemotePromiseHandler.java 2001/09/06 09:55:38 1.2
+++ RemotePromiseHandler.java 2001/10/02 01:29:17 1.3
@@ -34,12 +34,12 @@
private boolean myFreshFlag = true;
/**
- * Construct a new RemotePromiseHandler.
- *
- * @param connection The ProxyConnection to communicate via
- * @param pos The Imports or Questions map pos of the object
- */
- /*package*/ RemotePromiseHandler(ProxyConnection connection, int pos) {
+ * Construct a new RemotePromiseHandler.
+ *
+ * @param connection The CapTPConnection to communicate via
+ * @param pos The Imports or Questions map pos of the object
+ */
+ /*package*/ RemotePromiseHandler(CapTPConnection connection, int pos) {
super(connection, pos, null);
}
1.1 e/src/jsrc/net/captp/jcomm/CapTPConnection.java
Index: CapTPConnection.java
===================================================================
package net.captp.jcomm;
/*
The contents of this file are subject to the Electric Communities E Open
Source Code License Version 1.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.communities.com/EL/.
Software distributed under the License is distributed on an "AS IS" basis,
WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for
the specific language governing rights and limitations under the License.
The Original Code is the Distributed E Language Implementation, released
July 20, 1998.
The Initial Developer of the Original Code is Electric Communities.
Copyright (C) 1998 Electric Communities. All Rights Reserved.
Contributor(s): ______________________________________.
*/
import net.captp.tables.AnswersTable;
import net.captp.tables.ExportsTable;
import net.captp.tables.ImportsTable;
import net.captp.tables.NearGiftTable;
import net.captp.tables.PromiseGiftTable;
import net.captp.tables.QuestionsTable;
import net.captp.tables.SwissTable;
import net.vattp.data.VatTPConnection;
import net.vattp.data.Msg;
import net.vattp.data.MsgHandler;
import net.vattp.security.ESecureRandom;
import org.erights.e.develop.assertion.Assertion;
import org.erights.e.develop.exception.ExceptionMgr;
import org.erights.e.develop.exception.ThrowableSugar;
import org.erights.e.develop.trace.Trace;
import org.erights.e.elib.prim.E;
import org.erights.e.elib.ref.ProxyResolver;
import org.erights.e.elib.ref.Ref;
import org.erights.e.elib.ref.Resolver;
import org.erights.e.elib.sealing.Brand;
import org.erights.e.elib.sealing.SealedBox;
import org.erights.e.elib.sealing.Sealer;
import org.erights.e.elib.sealing.Unsealer;
import org.erights.e.elib.serial.Serializer;
import org.erights.e.elib.serial.Unserializer;
import org.erights.e.elib.tables.ConstList;
import org.erights.e.elib.util.OneArgFunc;
import org.erights.e.meta.java.math.BigIntegerSugar;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OptionalDataException;
import java.math.BigInteger;
/**
* Object which manages a CapTP protocol connection to a remote
* vat and which will receive incoming CapTP protocol messages.<p>
*
* The processMessage(byte[] message, VatTPConnection) method will be
* called when a message of type E_MSG is received from the remote vat.<p>
*
* CapTPConnection manages four tables for 2-vat interactions:
* <UL>
* <LI>Questions: Handlers for remote objects we requested </LI>
* <LI>Imports: Handlers for remote objects exported to us unsolicited </LI>
* <LI>Answers: Answers for remote Questions </LI>
* <LI>Exports: local objects matching remote Imports </LI>
* </UL>
*
* The four tables can be arranged into a matrix: <pre>
*
* | we generate pos | remote generates pos
* | |
* --------------------+---------------------+----------------------
* outgoing remote ref | |
* we have handler | Questions | Imports
* remote has object | - | +
* --------------------+---------------------+----------------------
* incoming remote ref | |
* we have object | Exports | Answers
* remote has handler | + | -
* --------------------+---------------------+----------------------
* </pre>
*
* When referencing an object or handler by pos number alone
* (independent of what table it should be looked up in), we
* distinguish the table by the sign of pos: Questions and Answers
* entries are referenced using a negative pos, while Imports and
* Exports entries use positive pos. The sign of pos is
* determined by the following rule: when pos is allocated by the
* outgoing side, it negative; when pos is allocated by the
* incoming side, pos is positive. The usefulness of this scheme
* is that whenever we know one axis of this table (which we always
* know from context), we can be determine the other axis by
* looking at the sign of pos. <p>
*
* @author Chip Morningstar
* @author Bill Frantz
* @author Mark Miller
*/
/*package*/ class CapTPConnection implements MsgHandler {
/**
* Queue delivery of message, no answer expected. <p>
*
* Start counting at 8 since the previous version of the protocol
* (0.8.9) stopped at 7.
*/
static private final byte DELIVER_ONLY_OP = 8;
/**
* Queue delivery of message, resolve answer to the outcome.
* <p>
* DELIVER_OP as 9 had a different argument order
*/
static private final byte DELIVER_OP = 13;
/**
* An import on the other has gone away dropping some number of
* wireCounts. Clean up the export table.
*/
static private final byte GC_EXPORT_OP = 10;
/**
* A question on the other has gone away. Clean up the answer table.
*/
static private final byte GC_ANSWER_OP = 11;
/**
* Please shut down the connection if you have no messages pending.
*/
static private final byte SHUTDOWN_OP = 12;
/** The CapTPMgr for our vat */
private CapTPMgr myProxyMgr;
/**
* Means of communication with our partner at the other end of the
* line.
*/
private VatTPConnection myDataConnection;
/**
* Number of things (XXX what's a "thing"?) which are currently holding
* the connection open
*/
private int myUseCount;
/** Number of messages sent on this connection */
private int mySendCount;
/** Number of messages received on this connection */
private int myReceiveCount;
/** Flag indicating we are in the midst of shutting down the connection */
private boolean myShuttingDownFlag;
/**
* If the connection died, anybody who talks to us will get this. <p>
*
* This should be null exactly when the connection is still alive.
*/
private Throwable myOptProblem;
/**
* Resolver of the promise used to buffer lookup requests made in midst
* of connection shutdown.
*/
private Resolver myOptBufferedLookups;
/** Generates new swiss numbers */
private ESecureRandom myEntropy;
/**
* The Questions table: these are handlers this end created and is
* expecting the other end to hook up to the relevent objects themselves
* (e.g., results of sendAll()s).
*/
private QuestionsTable myQuestions;
/**
* The Answers table: this is the counterpart to the Questions table
* at the other end of the connection. <p>
*
* Messages sent through a handler in the questions table will be
* delivered to the corresponding answer.
*/
private AnswersTable myAnswers;
/**
* The Imports table: these are handles to objects the other end
* exported. <p>
*
* We just hook up the handler in the specified location.
*/
private ImportsTable myImports;
/**
* The Exports table: this is the counterpart to the Imports table at
* the other end of the connection. <p>
*
* These are objects that have been exported from this end (i.e.,
* mentioned by us in the parameters of a message). The other end
* installs them in the same spot in its Imports table.
*/
private ExportsTable myExports;
/**
* At incoming position 0, for bringing about 3-vat introductions
* using nonces.
*/
private NonceLocator myLocalNonceLocator;
/**
* A remote reference to the other side's myLocalNonceLocator, at
* outgoing position 0, for bringing about 3-vat introductions
* using nonces.
*/
private Object myRemoteNonceLocator;
/**
* For bringing about 3-vat introductions of unresolved references
* using nonces.
*/
private PromiseGiftTable myPGifts;
/**
* For bringing about 3-vat introductions of Near references using
* nonces.
*/
private NearGiftTable myNGifts;
/**
* Constructor
*
* @param proxyMgr The CapTPMgr managing this connection
* @param dataConn The VatTP-level data connection to be associated
* with this CapTPConnection
*/
public CapTPConnection(CapTPMgr proxyMgr,
VatTPConnection dataConn,
ESecureRandom entropy)
throws IOException {
myProxyMgr = proxyMgr;
myDataConnection = dataConn;
myOptProblem = null;
myUseCount = 0;
myReceiveCount = 0;
mySendCount = 0;
myShuttingDownFlag = false;
myOptBufferedLookups = null;
/* We want to listen for E_MSG traffic */
myDataConnection.registerMsgHandler(Msg.E_MSG, this);
myEntropy = entropy;
/* Create a new set of empty tables */
myQuestions = new QuestionsTable();
myAnswers = new AnswersTable();
myImports = new ImportsTable();
myExports = new ExportsTable();
//XXX for now
LookupHandler lookupHandler = new LookupHandler(this);
myRemoteNonceLocator = lookupHandler.myResolver.getProxy();
myPGifts = new PromiseGiftTable(myRemoteNonceLocator);
myNGifts = new NearGiftTable();
myLocalNonceLocator = new NonceLocator(myPGifts,
myNGifts,
dataConn.getRemoteVatID(),
proxyMgr,
proxyMgr.getSwissTable());
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm("Create CapTPConnection " + this +
" mgr=" + proxyMgr +
" dataConn=" + dataConn +
" vat=%" + dataConn.getRemoteVatID() +
"/%" + dataConn.getLocalVatID());
}
}
/**
*
*/
/*package*/ PromiseGiftTable getPromiseGiftTable() { return myPGifts; }
/**
*
*/
/*package*/ NearGiftTable getNearGiftTable() { return myNGifts; }
/**
*
*/
/*package*/ Object getRemoteNonceLocator() {
return myRemoteNonceLocator;
}
/**
*
*/
/*package*/ Sealer getSealer() { return myProxyMgr.mySealer; }
/**
*
*/
/*package*/ Unsealer getUnsealer() { return myProxyMgr.myUnsealer; }
/**
* If non-null, this is a dead connection
*/
/*package*/ Throwable getOptProblem() {
return myOptProblem;
}
/**
* Return our identity.
*/
/*package*/ String localVatID() {
return myDataConnection.getLocalVatID();
}
/**
* Return the identity of the party to whom we are speaking.
*/
/*package*/ String remoteVatID() {
return myDataConnection.getRemoteVatID();
}
/**
*
*/
/*package*/ ConstList remoteSearchPath() {
return myDataConnection.getRemoteSearchPath();
}
/**
*
*/
/*package*/ ProxyResolver makeQuestion() {
int pos = -myQuestions.bind(null);
RemotePromiseHandler handler = new RemotePromiseHandler(this, pos);
ProxyResolver result = handler.myResolver;
myQuestions.put(-pos, result);
return result;
}
/**
* Drop this index in our own Questions table, and in the other vat's
* Answers table.
*/
/*package*/ void dropQuestion(int pos) {
myQuestions.free(-pos);
sendGCAnswerOp(pos);
}
/************************ Desc Creation *********************/
/**
*
*/
/*package*/ NewFarDesc newFarDesc(Object obj) {
SwissTable swiss = myProxyMgr.getSwissTable();
BigInteger id = swiss.getIdentity(obj);
BigInteger hash = BigIntegerSugar.cryptoHash(id);
return new NewFarDesc(myExports.newFarPos(obj), hash);
}
/**
*
*/
/*package*/ NewRemotePromiseDesc newRemotePromiseDesc(Object promise) {
int importPos = myExports.bind(promise);
BigInteger rdrBase = myEntropy.nextSwiss();
BigInteger rdrNum = BigIntegerSugar.cryptoHash(rdrBase);
BigInteger rdrHash = BigIntegerSugar.cryptoHash(rdrNum);
int rdrPos = -myQuestions.bind(null);
FarHandler handler = new FarHandler(this, rdrPos, rdrHash);
ProxyResolver rdrResolver = handler.myResolver;
myQuestions.put(-rdrPos, rdrResolver);
Object farRdr = rdrResolver.getProxy();
E.sendOnly(promise, "whenMoreResolved", farRdr);
return new NewRemotePromiseDesc(importPos, rdrPos, rdrBase);
}
/**
* Returns a NewFarDesc, NewRemotePromiseDesc, or an ImportDesc for
* exporting obj, which is assumed to be suitable for being in our
* exports table.
* <p>
* obj is assumes to be a Near reference to a PassByProxy object (actual
* or HONORARY), or eventual.
*/
/*package*/ ObjectRefDesc makeImportingDesc(Object obj) {
if (Ref.isNear(obj)) {
int index = myExports.indexFor(obj);
if (-1 == index) {
return newFarDesc(obj);
} else {
return new ImportDesc(index);
}
} else if (Ref.isEventual(obj)) {
return newRemotePromiseDesc(obj);
} else {
throw new RuntimeException("internal: Not exportable");
}
}
/**
* Figure out what kind of eventual reference 'ref' is, and return an
* appropriate descriptor for encoding it over the wire.
*/
/*package*/ ObjectRefDesc makeEventualDesc(Ref ref) {
Unsealer unsealer = getUnsealer();
Brand brand = unsealer.brand();
SealedBox optBox = ref.optMeta(brand);
RemoteHandler optHandler =
(RemoteHandler)unsealer.unseal(optBox, RemoteHandler.class);
if (null == optHandler) {
//a local promise
return newRemotePromiseDesc(ref);
} else {
if (this == optHandler.myConn) {
int pos = optHandler.getPos();
return new IncomingDesc(pos);
} else {
throw new RuntimeException("XXX 3vats not yet implemented");
}
}
}
/************************ Desc Messages *********************/
/**
* Dereferencing of a NewFarDesc. <p>
*
* On entry, importPos may be free, or may be allocated to an entry with
* a zero wireCount. (XXX we don't currently check the wirecount.)
* In the latter case, the entry is overwritten.
*
* @param importPos The import position at which a new FarRef should be
* created.
* @param swissHash The sameness identity of that FarRef.
* @return The newly created FarRef.
*/
/*package*/ Ref newFarRef(int importPos,
BigInteger swissHash)
{
FarHandler handler = new FarHandler(this, importPos, swissHash);
myImports.put(importPos, handler.myResolver);
return handler.myResolver.getProxy();
}
/**
* Dereferencing of a NewRemotePromiseDesc. <p>
*
* On entry, importPos may be free, or may be allocated to an entry with
* a zero wireCount. In the latter case, the entry is overwritten.
*
* @param importPos The import position at which the new RemotePromise
* should be created.
* @param rdrPos The answers position at which the Redirector of that
* new RemotePromise is made available.
* @param rdrBase The sameness identity of that Redirector must be the
* cryptohash of rdrBase.
* @return The newly created RemotePromise.
*/
/*package*/ Ref newRemotePromise(int importPos,
int rdrPos,
BigInteger rdrBase)
{
RemotePromiseHandler handler =
new RemotePromiseHandler(this, importPos);
ProxyResolver resolver = handler.myResolver;
myImports.put(importPos, resolver);
DelayedRedirector rdr = new DelayedRedirector(resolver);
myAnswers.put(-rdrPos, rdr, true);
myProxyMgr.getSwissTable().registerNewSwiss(rdr, rdrBase);
return resolver.getProxy();
}
/**
* Dereferencing of an ImportDesc. <p>
*
* On entry, importPos must be allocated, but may be allocated to an
* entry with a zero wireCount. <p>
*
* Return an imported Proxy, or its resolution, and, if it still has
* a handler (ie, it isn't resolved) increment its wire count.
*
* @param importPos The position of the import in the Imports table
* @return Whatever the resolution is of the Proxy in the appropriate
* table at importPos
*/
/*package*/ Ref getImport(int importPos) {
ProxyResolver pr = myImports.getProxyResolver(importPos);
RemoteHandler optHandler = (RemoteHandler)pr.optHandler();
if (null != optHandler) {
optHandler.countWireRef();
}
return pr.getProxy();
}
/**
* Dereferencing of an IncomingDesc
*
* Return an Exports or Answers table entry.
*
* @param incomingPos A positive pos refer to the Exports table, a
* negative one to the Answers table.
* @return Whatever object was in the appropriate table at
* incomingPos.
*/
/*package*/ Object getIncoming(int incomingPos) {
if (incomingPos > 0) {
return myExports.get(incomingPos);
} else if (incomingPos < 0) {
return myAnswers.get(-incomingPos);
} else {
return myLocalNonceLocator;
}
}
/**
* Dereferencing of a Promise3Desc
*
* @param searchPath hints to find the vat identified by vatID
* @param vatID The fingerprint of the public key of the vat hosting the
* object to be looked up.
* @param nonce Identifies the object in that vat's appropriate
* gift table.
* @param optFarVine Hold on to this until the object has been retrieved.
* @return A promise for the looked up object.
*/
/*package*/ Ref getLookup(ConstList searchPath,
String vatID,
BigInteger nonce,
Object optFarVine)
{
throw new RuntimeException("XXX getLookup/4 not yet implemented");
}
/**
* Dereferencing of a Far3Desc
*
* @param searchPath hints to find the vat identified by vatID
* @param vatID The fingerprint of the public key of the vat hosting the
* object to be looked up.
* @param nonce Identifies the object in that vat's appropriate
* gift table.
* @param swissHash Identity of object being looked up. getLookup
* returns a resolved reference with that identity. If
* it can't return a FarRef with that identity, then it
* returns a DisconnectedRef with that identity.
* @param optFarVine Hold on to this until the object has been retrieved.
* @return A promise for the looked up object.
*/
/*package*/ Ref getLookup(ConstList searchPath,
String vatID,
BigInteger nonce,
BigInteger swissHash,
Object optFarVine)
{
throw new RuntimeException("XXX getLookup/5 not yet implemented");
}
/***************************** receiving ************************/
/**
* Process an incoming message from the VatTPConnection.
*
* @param message The incoming message.
* @param dataConn The VatTPConnection on which the message arrived
*
* @see net.vattp.data.MsgHandler
*/
public void processMessage(byte[] message, VatTPConnection dataConn) {
try {
Assertion.test(dataConn == myDataConnection,
"message's VatTPConnection doesn't match");
if (myOptProblem != null) {
/* Just to be safe */
return;
}
if (myShuttingDownFlag) {
myShuttingDownFlag = false;
if (null != myOptBufferedLookups) {
submitLookups(myOptBufferedLookups);
myOptBufferedLookups = null;
}
}
ByteArrayInputStream bis = new ByteArrayInputStream(message);
/* The message type (the first byte) should *always* be E_MSG */
if ((byte)bis.read() != Msg.E_MSG) {
Assertion.fail
("CapTPConnection was handed a non-E_MSG message");
}
//Read the command byte *before* starting to interpret the stream
++myReceiveCount;
byte cmd = (byte)bis.read();
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"CapTPConnection " + this + " receive msg cmd=" + cmd);
}
OneArgFunc resolver = new CapTPResolver(this);
Unserializer uns = new Unserializer(bis, resolver);
receiveMsg(cmd, uns);
} catch (Throwable problem) {
/* They sent a badly formed message, trace it & ignore it */
if (Trace.captp.warning && Trace.ON) {
Trace.captp.warningm("ignoring", problem);
}
}
}
private void receiveMsg(byte cmd, Unserializer uns)
throws IOException, ClassNotFoundException, OptionalDataException {
switch (cmd) {
case DELIVER_ONLY_OP: {
int recipPos = uns.readInt();
String verb = ((String)uns.readUTF()).intern();
try {
Object[] args = (Object[])uns.readObject();
execDeliverOnlyOp(recipPos, verb, args);
} catch (Throwable problem) {
whyNoDeliverOnlyOp(recipPos, verb, problem);
}
break;
}
case DELIVER_OP: {
int answerPos = uns.readInt();
Object rdr = uns.readObject();
int recipPos = uns.readInt();
String verb = ((String)uns.readUTF()).intern();
try {
Object[] args = (Object[])uns.readObject();
execDeliverOp(answerPos, rdr,
recipPos, verb, args);
} catch (Throwable problem) {
whyNoDeliverOp(answerPos, rdr,
recipPos, verb, problem);
}
break;
}
case GC_EXPORT_OP: {
int exportPos = uns.readInt();
int wireCount = uns.readInt();
execGCExportOp(exportPos, wireCount);
break;
}
case GC_ANSWER_OP: {
int answerPos = uns.readInt();
execGCAnswerOp(answerPos);
break;
}
case SHUTDOWN_OP: {
int receivedCount = uns.readInt();
execShutdownOp(receivedCount);
break;
}
default: {
Assertion.fail("Invalid command byte " + cmd);
}
}
}
/**
* Pretty print the args of a message send, for debugging purposes.
*/
private String argsString(Object args[]) {
if (args.length == 0) {
return "[]";
}
String result = "[\n " + args[0];
for (int i = 1; i < args.length; ++i) {
result += ",\n " + args[i];
}
return result + "]";
}
/**
*
*/
private void execDeliverOnlyOp(int recipPos, String verb,
Object[] args)
{
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"exec DeliverOnlyOp(" + recipPos + ", " + verb + ", " +
argsString(args) + ")");
}
Object recip = getIncoming(recipPos);
E.sendAllOnly(recip, verb, args);
}
/**
*
*/
private void whyNoDeliverOnlyOp(int recipPos, String verb,
Throwable problem)
{
String msg = "whyNoDeliverOnlyOp(" + recipPos + ", " + verb + ", " +
problem + ")";
//Note: trace level is warning
if (Trace.captp.warning && Trace.ON) {
Trace.captp.warningm(msg, problem);
}
if (0 != recipPos || ! "traceRemote".equals(verb)) {
//Just locally report a failure to remotely report,
//in order to avoid a potential bounce cycle. All others
//are remotely reported as well.
E.sendOnly(myRemoteNonceLocator, "traceRemote", msg);
}
}
/**
*
*/
private void execDeliverOp(int answerPos, Object rdr,
int recipPos, String verb, Object[] args)
{
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"exec DeliverOp(" + answerPos + ", " + rdr + ",\n " +
recipPos + ", " + verb + ", " +
argsString(args) + ")");
}
Object recip = getIncoming(recipPos);
Ref answer = E.sendAll(recip, verb, args);
myAnswers.put(-answerPos, answer, true);
E.sendOnly(answer, "whenMoreResolved", rdr);
}
/**
*
*/
private void whyNoDeliverOp(int answerPos, Object rdr,
int recipPos, String verb, Throwable problem)
{
Ref broke = Ref.broken(problem);
//Note: trace level is warning
if (Trace.captp.warning && Trace.ON) {
Trace.captp.warningm(
"whyNoDeliverOp(" + answerPos + ", " + rdr + ",\n " +
recipPos + ", " + verb + ", ???)",
problem);
}
myAnswers.put(-answerPos, broke, true);
E.sendOnly(broke, "whenMoreResolved", rdr);
}
/**
*
*/
private void execGCExportOp(int exportPos, int wireCount) {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"exec GCExportOp(" + exportPos + ", " + wireCount + ")");
}
myExports.decr(exportPos, wireCount);
}
/**
*
*/
private void execGCAnswerOp(int answerPos) {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"exec GCAnswerOp(" + answerPos + ")");
}
myAnswers.free(-answerPos);
}
/**
* Receive a shutdown message from the other end. If we don't have any
* messages in flight, we shutdown the connection.
*/
private void execShutdownOp(int receivedCount) throws IOException {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"exec ShutdownOp(" + receivedCount + ")");
}
if (mySendCount <= receivedCount) {
killConnection(new IOException("received shutdown request"),
true);
}
}
/************************** sending ************************/
/**
*
*/
private Serializer makeMsg(ByteArrayOutputStream bos, byte cmd)
throws IOException {
bos.write(Msg.E_MSG);
bos.write(cmd);
OneArgFunc replacer = new CapTPReplacer(this);
return new Serializer(bos, replacer);
}
/**
* Send a message via our VatTPConnection.
*/
private void sendMsg(ByteArrayOutputStream bos, Serializer ser)
throws IOException {
ser.flush();
byte[] msg = bos.toByteArray();
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"CapTPConnection " + this + " sendMsg cmd=" + msg[1]);
}
++mySendCount;
myDataConnection.sendMsg(msg);
}
/**
*
*/
/*package*/ void sendDeliverOnlyOp(int recipPos,
String verb,
Object[] args)
{
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"send DeliverOnlyOp(" + recipPos + ", " + verb + ", " +
argsString(args) + ")");
}
if (null != myOptProblem) {
throw ExceptionMgr.asSafe(myOptProblem);
}
ByteArrayOutputStream bos;
Serializer ser;
try {
bos = new ByteArrayOutputStream();
ser = makeMsg(bos, DELIVER_ONLY_OP);
ser.writeInt(recipPos);
ser.writeUTF(verb);
} catch (Throwable problem) {
killConnection(problem, false);
throw ExceptionMgr.asSafe(problem);
}
try {
ser.writeObject(args);
} catch (Throwable problem) {
//don't kill the connection for a serialization problem
throw ExceptionMgr.asSafe(problem);
}
try {
sendMsg(bos, ser);
} catch (Throwable problem) {
killConnection(problem, false);
throw ExceptionMgr.asSafe(problem);
}
}
/**
*
*/
/*package*/ void sendDeliverOp(int answerPos,
Object rdr,
int recipPos,
String verb,
Object[] args)
{
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"send DeliverOp(" + answerPos + ", " + rdr + ",\n " +
recipPos + ", " + verb + ", " +
argsString(args) + ")");
}
if (null != myOptProblem) {
throw ExceptionMgr.asSafe(myOptProblem);
}
ByteArrayOutputStream bos;
Serializer ser;
try {
bos = new ByteArrayOutputStream();
ser = makeMsg(bos, DELIVER_OP);
ser.writeInt(answerPos);
ser.writeObject(rdr);
ser.writeInt(recipPos);
ser.writeUTF(verb);
} catch (Throwable problem) {
killConnection(problem, false);
throw ExceptionMgr.asSafe(problem);
}
try {
ser.writeObject(args);
} catch (Throwable problem) {
//don't kill the connection for a serialization problem
throw ExceptionMgr.asSafe(problem);
}
try {
sendMsg(bos, ser);
} catch (Throwable problem) {
killConnection(problem, false);
throw ExceptionMgr.asSafe(problem);
}
}
/**
*
*/
/*package*/ void sendGCExportOp(int exportPos, int wireCount) {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"send GCExportOp(" + exportPos + ", " + wireCount + ")");
}
if (null != myOptProblem) {
throw ExceptionMgr.asSafe(myOptProblem);
}
try {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
Serializer ser = makeMsg(bos, GC_EXPORT_OP);
ser.writeInt(exportPos);
ser.writeInt(wireCount);
sendMsg(bos, ser);
} catch (Throwable problem) {
killConnection(problem, false);
throw ExceptionMgr.asSafe(problem);
}
}
/**
*
*/
private void sendGCAnswerOp(int answerPos) {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"send GCAnswerOp(" + answerPos + ")");
}
if (null != myOptProblem) {
throw ExceptionMgr.asSafe(myOptProblem);
}
try {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
Serializer ser = makeMsg(bos, GC_ANSWER_OP);
ser.writeInt(answerPos);
sendMsg(bos, ser);
} catch (Throwable problem) {
killConnection(problem, false);
throw ExceptionMgr.asSafe(problem);
}
}
/**
*
*/
/*package*/ void sendShutdownOp(int receivedCount) {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm(
"send ShutdownOp(" + receivedCount + ")");
}
if (null != myOptProblem) {
throw ExceptionMgr.asSafe(myOptProblem);
}
try {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
Serializer ser = makeMsg(bos, SHUTDOWN_OP);
ser.writeInt(receivedCount);
sendMsg(bos, ser);
} catch (Throwable problem) {
killConnection(problem, false);
throw ExceptionMgr.asSafe(problem);
}
}
/**************** lookup, gc, & shutdown *****************/
/**
*
*/
/*package*/ void submitLookups(Resolver bufferedLookups) {
throw new RuntimeException("XXX submitLookups not yet implemented");
}
/**
* Decrement the use count. If it reaches 0, we have no references active
* over this connection and the connection can (and should) be shut down.
*/
private void decrementUseCount() throws IOException {
--myUseCount;
if (myUseCount <= 0) {
myShuttingDownFlag = true;
sendShutdownOp(myReceiveCount);
}
}
/**
* Receive notification that our VatTPConnection has died.
*
* @param dataConn The VatTPConnection object which has just died.
*
* @see net.vattp.data.MsgHandler
*/
public void connectionDead(VatTPConnection dataConn, Throwable problem) {
Assertion.test(dataConn == myDataConnection,
"dead VatTPConnection doesn't match");
killConnection(ThrowableSugar.backtrace(problem, "lost " + dataConn),
myShuttingDownFlag);
}
/**
* Terminate this connection in the case where something
* unrecoverable has gone wrong. <p>
*
* "A coward dies a thousand deaths, a hero dies but once." <p>
*
* Our error handling is cowardly -- if anything goes wrong during
* I/O we assume we are dead. Thus we can be caused to die more
* than once, but we need to actually die only once...
*
* @param problem What went wrong
*/
/*package*/ void killConnection(Throwable problem, boolean deliberate)
{
if (null == problem) {
problem = new NullPointerException("dying with a null problem");
}
if (deliberate) {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm("shutdown " + this, problem);
}
} else {
//at warning level because lost connections should be traced by
//default.
if (Trace.captp.warning && Trace.ON) {
Trace.captp.warningm("lost " + this, problem);
}
}
if (null == myOptProblem) { /* If this is the first time we've died... */
myOptProblem = problem;
myDataConnection.shutDownConnection(problem);
myQuestions.smash(problem);
myAnswers.smash(problem);
myImports.smash(problem);
myExports.smash(problem);
myPGifts.smash(problem);
myNGifts.smash(problem);
myQuestions= null;
myAnswers= null;
myImports= null;
myExports= null;
myPGifts= null;
myNGifts= null;
if (deliberate) {
//transfer buffered lookups to a newly spawned connection
myProxyMgr.connectionDead(myDataConnection,
myOptBufferedLookups);
} else {
//XXX smash buffered lookups
}
myOptBufferedLookups = null;
}
}
}
1.1 e/src/jsrc/net/captp/jcomm/CapTPMgr.java
Index: CapTPMgr.java
===================================================================
package net.captp.jcomm;
/*
The contents of this file are subject to the Electric Communities E Open
Source Code License Version 1.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.communities.com/EL/.
Software distributed under the License is distributed on an "AS IS" basis,
WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for
the specific language governing rights and limitations under the License.
The Original Code is the Distributed E Language Implementation, released
July 20, 1998.
The Initial Developer of the Original Code is Electric Communities.
Copyright (C) 1998 Electric Communities. All Rights Reserved.
Contributor(s): ______________________________________.
*/
import net.captp.tables.SwissTable;
import net.vattp.data.VatTPMgr;
import net.vattp.data.VatTPConnection;
import net.vattp.data.NetConfig;
import net.vattp.data.NewConnectionReactor;
import net.vattp.data.VatLocationLookup;
import net.vattp.security.ESecureRandom;
import org.erights.e.develop.exception.ExceptionMgr;
import org.erights.e.develop.trace.Trace;
import org.erights.e.elib.ref.Resolver;
import org.erights.e.elib.sealing.Brand;
import org.erights.e.elib.sealing.Sealer;
import org.erights.e.elib.sealing.Unsealer;
import org.erights.e.elib.tables.ConstList;
import org.erights.e.elib.tables.FlexMap;
import java.io.IOException;
import java.net.UnknownHostException;
import java.security.KeyPair;
/**
* Manage the interface between this vat and other vats. <p>
*
* The CapTPMgr acts as the focal point for managing the point-to-point
* connections between vats. It is the application layer's only interface with
* the low-level comm system.
*/
public class CapTPMgr implements NewConnectionReactor {
/** Registration table to use for map between ObjectIDs and objs */
private SwissTable mySwissTable;
/** map from DataConnections to ProxyConnections */
private FlexMap myProxyConnections = null;
/** The connections manager to talk to */
private VatTPMgr myConnMgr = null;
/** makes new unguessable numbers */
private ESecureRandom myEntropy;
/** for self recognition */
/*package*/ Sealer mySealer;
/*package*/ Unsealer myUnsealer;
/**
* Constructor
*
* @param vat The identity of the vat we are managing remote references
* into.
* @param netConfig Where to listen, register, and tell other to look for
* me.
* @param swissTable Associates swissNumbers with references.
* @param entropy Provides new unguessable numbers.
*/
public CapTPMgr(KeyPair identityKeys,
NetConfig netConfig,
SwissTable swissTable,
ESecureRandom entropy)
throws UnknownHostException, IOException {
myConnMgr = new VatTPMgr(identityKeys, netConfig);
myConnMgr.addNewConnectionReactor(this);
mySwissTable = swissTable;
myProxyConnections = FlexMap.fromTypes(VatTPConnection.class,
CapTPConnection.class);
myEntropy = entropy;
Object[] pair = Brand.pair("captp");
mySealer = (Sealer)pair[0];
myUnsealer = (Unsealer)pair[1];
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm("Create CapTPMgr " + this +
" connMgr=" + myConnMgr);
}
}
/**
* Do the work of setting up a new CapTPConnection, regardless of which
* end initiated the transaction. If there is already an existing
* connection to the indicated party, just use it.
*
* @param dataConn The VatTPConnection upon which we are communicating.
*
* @return The resulting new (or existing) CapTPConnection.
*/
private CapTPConnection arrangeProxyConnection(VatTPConnection dataConn)
throws IOException {
CapTPConnection proxyConn = null;
try {
/* First, look for an existing one... */
proxyConn = (CapTPConnection) myProxyConnections.get(dataConn);
} catch (IndexOutOfBoundsException ex) {
/* Well, that didn't work. Make a new one, then. */
proxyConn = new CapTPConnection(this, dataConn, myEntropy);
myProxyConnections.put(dataConn, proxyConn);
}
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm("CapTPMgr " + this +
" arrangeConnection " +
dataConn + "->" + proxyConn);
}
return proxyConn;
}
/**
* Be informed that one of our connections has died.
*
* @param connection The connection that died
* @param bufferedLookups SturdyRef lookup requests that were issued in
* the midst of orderly shutdown (these need to be resubmitted).
*/
void connectionDead(VatTPConnection connection, Resolver optBufferedLookups) {
myProxyConnections.removeKey(connection, true);
if (null == optBufferedLookups) {
return;
}
/* Get back into it, then */
try {
CapTPConnection proxyConn =
getOrMakeProxyConnection(connection.getRemoteSearchPath(),
connection.getRemoteVatID());
proxyConn.submitLookups(optBufferedLookups);
} catch (IOException ioe) {
optBufferedLookups.smash(ioe);
} catch (Throwable problem) {
optBufferedLookups.smash(problem);
throw ExceptionMgr.asSafe(problem);
}
}
/**
*
*/
/*package*/ SwissTable getSwissTable() { return mySwissTable; }
/**
* Notice that a new (inbound) data connection has appeared.
*
* @param connection The new VatTPConnection object.
*
* @see net.vattp.data.NewConnectionReactor
*/
public void reactToNewConnection(VatTPConnection connection) {
if (Trace.captp.debug && Trace.ON) {
Trace.captp.debugm("CapTPMgr " + this +
" reactToNewConnection " + connection +
" rvat=%" + connection.getRemoteVatID() +
" lvat=%" + connection.getLocalVatID());
}
try {
arrangeProxyConnection(connection);
} catch (IOException e) {
/* I guess it wasn't meant to be... */
}
}
/**
* If I've got a live CapTPConnection for that vatID, return it, else
* null. <p>
*
* This does *not* initiate a connection attempt.
*/
/*package*/ CapTPConnection optProxyConnection(String vatID)
throws IOException {
VatTPConnection optDataConn =
myConnMgr.optConnection(vatID);
if (null == optDataConn) {
return null;
} else {
return arrangeProxyConnection(optDataConn);
}
}
/**
* If vatID is me, return null, else get or make a live CapTPConnection
* for vatID. <p>
*/
/*package*/ CapTPConnection getOrMakeProxyConnection(ConstList searchPath,
String vatID)
throws IOException {
VatTPConnection optDataConn =
myConnMgr.getConnection(vatID, searchPath);
if (null == optDataConn) {
return null;
} else {
return arrangeProxyConnection(optDataConn);
}
}
/**
* Return the search path to our own vat.
*/
public ConstList searchPath() {
return myConnMgr.searchPath();
}
/**
* Return the post-on-the-air configuration parameters
*/
public NetConfig getNetConfig() {
return myConnMgr.getNetConfig();
}
/**
* Set the object which will respond to vat location queries on our
* listen port.
*
* @param vls The VLS object
*/
public void setVatLocationLookup(VatLocationLookup vls) {
myConnMgr.setVatLocationLookup(vls);
}
}
1.2 +1 -1 e/src/jsrc/net/vattp/data/AuthSecrets.java
Index: AuthSecrets.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/AuthSecrets.java,v
retrieving revision 1.1
retrieving revision 1.2
diff -u -r1.1 -r1.2
--- AuthSecrets.java 2000/12/21 22:15:18 1.1
+++ AuthSecrets.java 2001/10/02 01:29:18 1.2
@@ -22,7 +22,7 @@
/**
* Structure to hold a change in protocol request in the queue
- * between the DataConnection and the SendThread
+ * between the VatTPConnection and the SendThread
*
* @author Bill Frantz
*/
1.4 +22 -22 e/src/jsrc/net/vattp/data/DataCommThunk.java
Index: DataCommThunk.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/DataCommThunk.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- DataCommThunk.java 2001/09/06 09:55:39 1.3
+++ DataCommThunk.java 2001/10/02 01:29:18 1.4
@@ -27,7 +27,7 @@
/**
* This class is used by SendThread, RecvThread and ListenThread
* to schedule calls to classes inside the vat
- * (in DataPath or ConnectionsMgr). The call is
+ * (in DataPath or VatTPMgr). The call is
* scheduled by instantitating an instance of this object through
* the constructor appropriate for the method to be called. Then
* that instance is passed to Runner's now(...) method.
@@ -47,7 +47,7 @@
int type = 0;
private DataPath myDataPath; // For Send/RecvThread calls
- private ConnectionsMgr myConnMgr; // For ListenThread calls
+ private VatTPMgr myConnMgr; // For ListenThread calls
private Throwable myException;
private int myWrittenLength;
private int myWrittenCount;
@@ -60,26 +60,26 @@
private String myAddress;
/**
- * Construct a DataCommThunk to call noticeProblem() in a
- * ConnectionsMgr object.
- *
- * @param connMgr the ConnectionsMgr object to invoke.
- * @param e the Exception object describing the problem.
- */
- /*package*/ DataCommThunk(ConnectionsMgr connMgr, Throwable e) {
+ * Construct a DataCommThunk to call noticeProblem() in a
+ * VatTPMgr object.
+ *
+ * @param connMgr the VatTPMgr object to invoke.
+ * @param e the Exception object describing the problem.
+ */
+ /*package*/ DataCommThunk(VatTPMgr connMgr, Throwable e) {
myConnMgr = connMgr;
myException = e;
type = NOTICE_PROBLEM_CM;
}
/**
- * Construct a DataCommThunk to call newInboundSocket in a
- * ConnectionsMgr object.
- *
- * @param connMgr the DataPath object to invoke.
- * @param The byte array which is the message.
- */
- /*package*/ DataCommThunk(ConnectionsMgr connMgr, Socket socket) {
+ * Construct a DataCommThunk to call newInboundSocket in a
+ * VatTPMgr object.
+ *
+ * @param connMgr the DataPath object to invoke.
+ * @param The byte array which is the message.
+ */
+ /*package*/ DataCommThunk(VatTPMgr connMgr, Socket socket) {
myConnMgr = connMgr;
mySocket = socket;
type = INCOMING_SOCKET;
@@ -159,12 +159,12 @@
}
/**
- * Construct a DataCommThunk to call newInboundSocket in a
- * ConnectionsMgr object.
- *
- * @param conn the DataPath object to invoke.
- * @param The byte array which is the message.
- */
+ * Construct a DataCommThunk to call newInboundSocket in a
+ * VatTPMgr object.
+ *
+ * @param conn the DataPath object to invoke.
+ * @param The byte array which is the message.
+ */
/*package*/ DataCommThunk(DataPath conn, RecvThread recv,
NetAddr remoteNetAddr,
NetAddr localNetAddr) {
1.6 +126 -126 e/src/jsrc/net/vattp/data/DataPath.java
Index: DataPath.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/DataPath.java,v
retrieving revision 1.5
retrieving revision 1.6
diff -u -r1.5 -r1.6
--- DataPath.java 2001/09/14 20:54:04 1.5
+++ DataPath.java 2001/10/02 01:29:18 1.6
@@ -64,14 +64,14 @@
* received within . */
static private final byte[] thePingMsg = {Msg.PING};
- /** The DataConnection we report status changes to. It is connected late
+ /** The VatTPConnection we report status changes to. It is connected late
* in the startup protocol for an incoming connection */
- private DataConnection /*nullOK*/ myDataConnection;
+ private VatTPConnection /*nullOK*/ myDataConnection;
- /** The ConnectionsMgr we report the remote identity to for an
+ /** The VatTPMgr we report the remote identity to for an
* incoming connection. Null for outgoing connections unless they are
* connectToVatAt connections. */
- private ConnectionsMgr /*nullOK*/ myConnMgr;
+ private VatTPMgr /*nullOK*/ myConnMgr;
/** Array of MsgHandlers indexed by message type */
private MsgHandler[] myMsgHandlers = new MsgHandler[Msg.HIGH_MSG_TYPE+1];
@@ -80,7 +80,7 @@
* during construction, and set to null in shutDownPath. */
private SynchQueue myWriter;
- /** The RecvThread receiving messages for this DataConnection */
+ /** The RecvThread receiving messages for this VatTPConnection */
private RecvThread myRecvThread;
/** The SendThread we use for output. We keep this reference
@@ -137,7 +137,7 @@
private String myLocalFlattenedSearchPath;
/** The search path the other end provided on an incoming connection.
- * We save this and give it to the DataConnection at the end of the
+ * We save this and give it to the VatTPConnection at the end of the
* startup protocol so we have a search path to use to resume the
* connection after it has been suspended. */
private String mySavedFlattenedSearchPath;
@@ -182,26 +182,26 @@
private ConditionLock myEmbargoLock = new ConditionLock();
/**
- * Make a new DataPath for an incoming connection
- *
- * @param connection The DataConnection to notify of status
- * changes.
- *
- * @param socket The Socket to the TCP connection.
- *
- * @param identityKeys is the KeyPair which defines the identity of this
- * vat.
- *
- * @param runner is the Runner whose thread we use to synchronize calls
- * into the DataPath object from the SendThread and ReceiveThread
- *
- * @param localFlattenedSearchPath Is the search path we publish for
- * references to us, flattened into a string.
- *
- * @param localSearchPath Is the search path we publish for references
- * to us, with each element a separate String in an array.
- */
- /*package*/ DataPath(ConnectionsMgr connMgr,
+ * Make a new DataPath for an incoming connection
+ *
+ * @param connection The VatTPConnection to notify of status
+ * changes.
+ *
+ * @param socket The Socket to the TCP connection.
+ *
+ * @param identityKeys is the KeyPair which defines the identity of this
+ * vat.
+ *
+ * @param runner is the Runner whose thread we use to synchronize calls
+ * into the DataPath object from the SendThread and ReceiveThread
+ *
+ * @param localFlattenedSearchPath Is the search path we publish for
+ * references to us, flattened into a string.
+ *
+ * @param localSearchPath Is the search path we publish for references
+ * to us, with each element a separate String in an array.
+ */
+ /*package*/ DataPath(VatTPMgr connMgr,
Socket tcpConnection,
KeyPair identityKeys,
String localVatID,
@@ -228,48 +228,48 @@
}
/**
- * Make a new DataPath
- *
- * @param connMgr is the ConnectionsMgr if this DataPath is supporting
- * a "connectToVatAt" operation. Otherwise it is null.
- *
- * @param connection The DataConnection to notify of status
- * changes.
- *
- * @param remoteVatID the vatID of the remote vat
- *
- * @param remoteAddr is the dotted IP address or DNS name and port number
- * of the place to try for the remote vat.
- *
- * @param addressesTried is a Hashtable of the InetAddresses already
- * tried to locate the vat which failed due to host,
- * rather than port related problems, and are not to be
- * tried again.
- *
- * @param identityKeys is the KeyPair which defines the identity of this
- * vat.
- *
- * @param runner is the Runner whose thread we use to synchronize calls
- * into the DataPath object from the SendThread and ReceiveThread
- *
- * @param outgoingSuspendID Is the Suspend ID we are to send to the
- * remote vat for resuming a connection, or null for a new connection.
- *
- * @param macKey is the key used for message authentication for resume,
- * or null for a new connection.
- *
- * @param protocolParms is the AuthSecrets object which has the resume
- * state for protocol versions an protocol parameters for resuming a
- * connection, or null for a new connection.
- *
- * @param localFlattenedSearchPath Is the search path we publish for
- * references to us, flattened into a string.
- *
- * @param localSearchPath Is the search path we publish for references
- * to us, with each element a separate String in an array.
- */
- /*package*/ DataPath(ConnectionsMgr /*nullOK*/ connMgr,
- DataConnection /*nullOK*/ connection,
+ * Make a new DataPath
+ *
+ * @param connMgr is the VatTPMgr if this DataPath is supporting
+ * a "connectToVatAt" operation. Otherwise it is null.
+ *
+ * @param connection The VatTPConnection to notify of status
+ * changes.
+ *
+ * @param remoteVatID the vatID of the remote vat
+ *
+ * @param remoteAddr is the dotted IP address or DNS name and port number
+ * of the place to try for the remote vat.
+ *
+ * @param addressesTried is a Hashtable of the InetAddresses already
+ * tried to locate the vat which failed due to host,
+ * rather than port related problems, and are not to be
+ * tried again.
+ *
+ * @param identityKeys is the KeyPair which defines the identity of this
+ * vat.
+ *
+ * @param runner is the Runner whose thread we use to synchronize calls
+ * into the DataPath object from the SendThread and ReceiveThread
+ *
+ * @param outgoingSuspendID Is the Suspend ID we are to send to the
+ * remote vat for resuming a connection, or null for a new connection.
+ *
+ * @param macKey is the key used for message authentication for resume,
+ * or null for a new connection.
+ *
+ * @param protocolParms is the AuthSecrets object which has the resume
+ * state for protocol versions an protocol parameters for resuming a
+ * connection, or null for a new connection.
+ *
+ * @param localFlattenedSearchPath Is the search path we publish for
+ * references to us, flattened into a string.
+ *
+ * @param localSearchPath Is the search path we publish for references
+ * to us, with each element a separate String in an array.
+ */
+ /*package*/ DataPath(VatTPMgr /*nullOK*/ connMgr,
+ VatTPConnection /*nullOK*/ connection,
String remoteVatID,
String remoteAddr,
Hashtable addressesTried,
@@ -418,16 +418,16 @@
}
/**
- * Connect this DataPath to a DataConnection. Used for incoming sockets
- * and anonymous outgoing sockets to connect the DataPath after the
- * remote end has been identified.
- *
- * @param connection is the DataConnection object to connect to.
- * @param protocolParms is the protocol versions and parameters
- * to use for resuming a connection, or null for none.
- * @return is the flattened search path to use to find the remote end.
- */
- /*package*/ String connectConnection(DataConnection connection,
+ * Connect this DataPath to a VatTPConnection. Used for incoming sockets
+ * and anonymous outgoing sockets to connect the DataPath after the
+ * remote end has been identified.
+ *
+ * @param connection is the VatTPConnection object to connect to.
+ * @param protocolParms is the protocol versions and parameters
+ * to use for resuming a connection, or null for none.
+ * @return is the flattened search path to use to find the remote end.
+ */
+ /*package*/ String connectConnection(VatTPConnection connection,
AuthSecrets protocolParms)
{
myDataConnection = connection;
@@ -443,7 +443,7 @@
/**
* Receive notification that the connection is dead. Either the connection
- * has been shut down with DataConnection.shutDownConnection, the underlying
+ * has been shut down with VatTPConnection.shutDownConnection, the underlying
* TCP connection has failed, a bad MAC has been received on the connection,
* or an internal error has caused the connection to fail.
* <p>
@@ -451,11 +451,11 @@
* receive one connectionDead notification for each message type for which
* it is registered.
*
- * @param connection The DataConnection object which has just died.
+ * @param connection The VatTPConnection object which has just died.
* For this use, it may be null.
* @param reason is a Throwable describing why the connection was shut down.
*/
- public void connectionDead(DataConnection connection, Throwable reason) {
+ public void connectionDead(VatTPConnection connection, Throwable reason) {
//Do nothing - part of the MsgHandler interface.
}
@@ -505,7 +505,7 @@
Trace.comm.verbosem("enqueue " + message);
}
//If the path has been shutdown, myWriter will be null. If a client
- //is trying to write, that should be caught by the DataConnection.
+ //is trying to write, that should be caught by the VatTPConnection.
//This means that the writer is StartUpProtocol of DataPath. If the
//path is going away, their messages can be trashed.
if (null == myWriter) return;
@@ -548,14 +548,14 @@
}
/**
- * Identify the remote vat's vatID for an incoming connection.
- *
- * @param remoteVatID the VatID of the remote vat.
- * @param remoteSearchPath the search path to use to find the remote vat
- * in array form.
- * @return the code received from
- ConnectionsMgr.newIncomingConnectionIdentified
- */
+ * Identify the remote vat's vatID for an incoming connection.
+ *
+ * @param remoteVatID the VatID of the remote vat.
+ * @param remoteSearchPath the search path to use to find the remote vat
+ * in array form.
+ * @return the code received from
+ VatTPMgr.newIncomingConnectionIdentified
+ */
/*package*/ int identifyIncoming(String localVatID,
String remoteVatID,
String remoteSearchPath)
@@ -563,7 +563,7 @@
{
myRemoteVatID = remoteVatID;
- //Save the search path to give to the DataConnection when startup
+ //Save the search path to give to the VatTPConnection when startup
//is finished. It will need the search path to resume a suspended
//connection.
mySavedFlattenedSearchPath = remoteSearchPath;
@@ -595,7 +595,7 @@
{
myRemoteVatID = remoteVatID;
- //Save the search path to give to the DataConnection when startup
+ //Save the search path to give to the VatTPConnection when startup
//is finished. It will need the search path to resume a suspended
//connection.
mySavedFlattenedSearchPath = remoteFlattenedSearchPath;
@@ -603,7 +603,7 @@
remoteVatID,
localVatID,
false);
- if (ConnectionsMgr.LIVES_DUP == s) {
+ if (VatTPMgr.LIVES_DUP == s) {
return "Connection alreay exists or in progress";
}
return myDataConnection.getOutgoingSuspendID();
@@ -732,20 +732,20 @@
}
//Methods from the MsgHandler interface.
/**
- * Process an incoming message from the DataConnection.
+ * Process an incoming message from the VatTPConnection.
*
* @param message is the incoming message. The first byte (message[0])
* is the message type (see class Msg). A handler which is registered
* for more than one message can use the initial byte to determine what
* is the message type of the current message.
*
- * @param connection is the DataConnection object on which the the message
+ * @param connection is the VatTPConnection object on which the the message
* arrived. For this use, it may be null.
*
* @see Msg
- * @see DataConnection
+ * @see VatTPConnection
*/
- public void processMessage(byte[] message, DataConnection connection) {
+ public void processMessage(byte[] message, VatTPConnection connection) {
if (Msg.PING == message[0]) { // got a ping, send pong
//No IOException?
// try {
@@ -766,30 +766,30 @@
}
/**
- * Register a MsgHandler for a particular message type. It will be called
- * whenever a new message of the specified type arrives. It will also
- * be called when the connection dies.
- *
- * @param msgType is the message type. It must be chosen from the types
- * defined in the definition class Msg. It is an error to register more
- * than one handler for a message type.
- * It is a fatal error if the msgType is out of range.
- * It is a fatal error to register a handler after the
- * reactToNewConnection method of the object registered with the
- * ConnectionsMgr has returned.
- *
- * @param handler is an object which implements the MsgHandler interface.
- * It will be called with processMessage when a message
- * of msgType is received from the remote vat. It will be called with
- * connectionDead when this connection dies.
- *
- * @exception IOException is thrown if
- * there is already a handler registered for this message type.
- *
- * @see Msg
- * @see MsgHandler
- * @see ConnectionsMgr
- */
+ * Register a MsgHandler for a particular message type. It will be called
+ * whenever a new message of the specified type arrives. It will also
+ * be called when the connection dies.
+ *
+ * @param msgType is the message type. It must be chosen from the types
+ * defined in the definition class Msg. It is an error to register more
+ * than one handler for a message type.
+ * It is a fatal error if the msgType is out of range.
+ * It is a fatal error to register a handler after the
+ * reactToNewConnection method of the object registered with the
+ * VatTPMgr has returned.
+ *
+ * @param handler is an object which implements the MsgHandler interface.
+ * It will be called with processMessage when a message
+ * of msgType is received from the remote vat. It will be called with
+ * connectionDead when this connection dies.
+ *
+ * @exception IOException is thrown if
+ * there is already a handler registered for this message type.
+ *
+ * @see Msg
+ * @see MsgHandler
+ * @see VatTPMgr
+ */
/*package*/ void registerMsgHandler (int msgType, MsgHandler handler)
throws IOException {
@@ -961,11 +961,11 @@
}
/**
- * Stop the start up protocol on this DataPath. This method is
- * called by the ConnectionsMgr when it determines that there are
- * two connections being built between this vat and another vat, and
- * that this connection is the one that should be abandonded.
- */
+ * Stop the start up protocol on this DataPath. This method is
+ * called by the VatTPMgr when it determines that there are
+ * two connections being built between this vat and another vat, and
+ * that this connection is the one that should be abandonded.
+ */
/*package*/ void stopStartUpProtocol() {
myStartUpProtocol.stopStartUpProtocol();
myDataConnection = null; // We've been disowned
1.4 +14 -14 e/src/jsrc/net/vattp/data/ListenThread.java
Index: ListenThread.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/ListenThread.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- ListenThread.java 2001/09/06 09:55:39 1.3
+++ ListenThread.java 2001/10/02 01:29:18 1.4
@@ -48,10 +48,10 @@
private InetAddress myOptIP = null;
private ServerSocket myListenServerSocket;
- /** The ConnectionsMgr to report new Sockets to */
- private ConnectionsMgr myConnMgr;
+ /** The VatTPMgr to report new Sockets to */
+ private VatTPMgr myConnMgr;
- /** A runner for synchronization when calling the ConnectionsMgr */
+ /** A runner for synchronization when calling the VatTPMgr */
private Runner myRunner;
/** An object to synchronize communication with the user thread */
@@ -59,16 +59,16 @@
private boolean mySuspended = false;
/**
- * Construct a new object to run in its own thread and listen on a
- * ServerSocket for new incoming connections.
- *
- * @param optLocalAddress The local address we will listen on.
- * @param connMgr the ConnectionsMgr to be notified about
- * interesting events (new connections and errors) that happen while
- * listening.
- */
+ * Construct a new object to run in its own thread and listen on a
+ * ServerSocket for new incoming connections.
+ *
+ * @param optLocalAddress The local address we will listen on.
+ * @param connMgr the VatTPMgr to be notified about
+ * interesting events (new connections and errors) that happen while
+ * listening.
+ */
/*package*/ ListenThread(String optLocalAddress,
- ConnectionsMgr connMgr,
+ VatTPMgr connMgr,
Runner runner)
throws UnknownHostException, IOException {
setDaemon(true);
@@ -89,7 +89,7 @@
}
/**
- * Call a method in the ConnectionsMgr
+ * Call a method in the VatTPMgr
*
* @param thunk a Thunk that will perform the call. The thunk will
* be called after the Vat lock is obtained.
@@ -181,7 +181,7 @@
/**
* Create a new connection to handle the new socket.
*
- * Tell the ConnectionsMgr about the new Socket.
+ * Tell the VatTPMgr about the new Socket.
*/
private void setupNewConnection(Socket clientSocket) {
callMgr(new DataCommThunk(myConnMgr, clientSocket));
1.4 +2 -2 e/src/jsrc/net/vattp/data/Msg.java
Index: Msg.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/Msg.java,v
retrieving revision 1.3
retrieving revision 1.4
diff -u -r1.3 -r1.4
--- Msg.java 2001/09/06 09:55:39 1.3
+++ Msg.java 2001/10/02 01:29:18 1.4
@@ -34,7 +34,7 @@
static /*package*/ final String[] Version = {"E0", "E0"};
-// Valid initial bytes in calls to DataConnection.sendMsg(...)
+// Valid initial bytes in calls to VatTPConnection.sendMsg(...)
// Value zero is not defined as an error check to make sure callers
// actually assign something to the first byte of the message.
@@ -75,7 +75,7 @@
/** Maximum length message permitted on a connection. This limit is
- * enforced by the DataConnection object.*/
+ * enforced by the VatTPConnection object.*/
static public final int MAX_OUTBOUND_MSG_LENGTH = 1024*1024;
/** Maximum length message we will receive. This length is set longer
1.3 +8 -8 e/src/jsrc/net/vattp/data/MsgHandler.java
Index: MsgHandler.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/MsgHandler.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- MsgHandler.java 2001/09/06 09:55:39 1.2
+++ MsgHandler.java 2001/10/02 01:29:18 1.3
@@ -25,7 +25,7 @@
* Interface to receive incoming messages.
*
* An object which implements the MsgHandler interface can be registered
- * with a DataConnection object to handle incoming messages.
+ * with a VatTPConnection object to handle incoming messages.
* It's processMessage() method will be called when a message
* of msgType is received from the remote vat.
* It's connectionDead() method will be called when the connection dies.
@@ -37,7 +37,7 @@
/**
* Receive notification that the connection is dead. Either the connection
- * has been shut down with DataConnection.shutDownConnection, the underlying
+ * has been shut down with VatTPConnection.shutDownConnection, the underlying
* TCP connection has failed, a bad MAC has been received on the connection,
* or an internal error has caused the connection to fail.
* <p>
@@ -45,24 +45,24 @@
* receive one connectionDead notification for each message type for which
* it is registered.
*
- * @param connection The DataConnection object which has just died.
+ * @param connection The VatTPConnection object which has just died.
* @param reason is a Throwable which describes why the connection died.
*/
- public void connectionDead(DataConnection connection, Throwable reason);
+ public void connectionDead(VatTPConnection connection, Throwable reason);
/**
- * Process an incoming message from the DataConnection.
+ * Process an incoming message from the VatTPConnection.
*
* @param message is the incoming message. The first byte (message[0])
* is the message type (see class Msg). A handler which is registered
* for more than one message can use the initial byte to determine what
* is the message type of the current message.
*
- * @param connection is the DataConnection object on which the the message
+ * @param connection is the VatTPConnection object on which the the message
* arrived.
*
* @see Msg
- * @see DataConnection
+ * @see VatTPConnection
*/
- public void processMessage(byte[] message, DataConnection connection);
+ public void processMessage(byte[] message, VatTPConnection connection);
}
1.7 +1 -1 e/src/jsrc/net/vattp/data/NetConfig.java
Index: NetConfig.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/NetConfig.java,v
retrieving revision 1.6
retrieving revision 1.7
diff -u -r1.6 -r1.7
--- NetConfig.java 2001/09/07 05:49:20 1.6
+++ NetConfig.java 2001/10/02 01:29:18 1.7
@@ -33,7 +33,7 @@
/**
* Configuration parameters for initializing or re-initializing a
- * ConnectionsMgr
+ * VatTPMgr
*
* @author <a href="mailto:markm@erights.org">Mark S. Miller</a>
*/
1.3 +3 -3 e/src/jsrc/net/vattp/data/NewConnectionReactor.java
Index: NewConnectionReactor.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/NewConnectionReactor.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- NewConnectionReactor.java 2001/09/06 09:55:39 1.2
+++ NewConnectionReactor.java 2001/10/02 01:29:18 1.3
@@ -29,13 +29,13 @@
/**
* React to the arrival of a new connection. If the object implementing
* NewConnectionReactor is registered through addNewConnectionReactor
- * on a ConnectionsMgr object, this invocation of it
+ * on a VatTPMgr object, this invocation of it
* is expected to connect up MsgHandlers for all the messages processed
* by higher layers in the protocol.
* This notification is presented for all DataConnections, whether the
* connection originated on this machine or on the remote machine.
*
- * @param connection is the new DataConnection object.
+ * @param connection is the new VatTPConnection object.
*/
- public void reactToNewConnection(DataConnection connection);
+ public void reactToNewConnection(VatTPConnection connection);
}
1.6 +2 -2 e/src/jsrc/net/vattp/data/RecvThread.java
Index: RecvThread.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/RecvThread.java,v
retrieving revision 1.5
retrieving revision 1.6
diff -u -r1.5 -r1.6
--- RecvThread.java 2001/09/06 09:55:39 1.5
+++ RecvThread.java 2001/10/02 01:29:18 1.6
@@ -41,7 +41,7 @@
/**
* A thread to read bytes from an TCP InputStream and pass them to a
- * DataConnection.
+ * VatTPConnection.
* <p>
* Created by SendThread when a Socket becomes available. This
* will be either after an outgoing connection succeeds, or after an
@@ -121,7 +121,7 @@
myRunner = runner;
}
/**
- * Call a method in our ConnectionsMgr
+ * Call a method in our VatTPMgr
*
* @param thunk a Thunk that will perform the call. The thunk will
* be called after the Vat lock is obtained.
1.5 +11 -11 e/src/jsrc/net/vattp/data/StartUpProtocol.java
Index: StartUpProtocol.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/StartUpProtocol.java,v
retrieving revision 1.4
retrieving revision 1.5
diff -u -r1.4 -r1.5
--- StartUpProtocol.java 2001/09/14 20:54:04 1.4
+++ StartUpProtocol.java 2001/10/02 01:29:18 1.5
@@ -53,7 +53,7 @@
/**
- * This class is used by DataConnection to perform the startup
+ * This class is used by VatTPConnection to perform the startup
* protocol.
* @author Bill Frantz
*/
@@ -397,7 +397,7 @@
/**
* Process a connection failure
*/
- public void connectionDead(DataConnection /*nullOK*/willBeNull, Throwable reason) {
+ public void connectionDead(VatTPConnection /*nullOK*/willBeNull, Throwable reason) {
// Everything should be handled by the DataPath
}
/**
@@ -491,17 +491,17 @@
myRemoteVatID,
remoteSearchPath);
switch (i) {
- case ConnectionsMgr.LIVES_CONTINUE: {
+ case VatTPMgr.LIVES_CONTINUE: {
myState = ST_INCOMING_EXPECT_GO;
sendReplyInfo();
break;
}
- case ConnectionsMgr.LIVES_DUP: {
+ case VatTPMgr.LIVES_DUP: {
myDataPath.stopStartUpProtocol();
startupError(TOK_DUP, "Crossed connections");
break;
}
- case ConnectionsMgr.LIVES_NOTIFY: {
+ case VatTPMgr.LIVES_NOTIFY: {
myState = ST_INCOMING_EXPECT_GO;
sendYouChose();
break;
@@ -1031,7 +1031,7 @@
* @param packetArray The startup protocol message to process.
*/
public void processMessage(byte packetArray[],
- DataConnection /*nullOK*/willBeNull) {
+ VatTPConnection /*nullOK*/willBeNull) {
ByteArrayInputStream byteArrayIn = new ByteArrayInputStream(packetArray);
DataInputStream packetIn = new DataInputStream(byteArrayIn);
Exception problem = null;
@@ -1485,11 +1485,11 @@
}
// Method for use by DataPath
/**
- * Stop the start up protocol. This method is called by the DataPath
- * relaying a call from the ConnectionsMgr when it determines that
- * there are two connections being built between this vat and another vat,
- * andthat this connection is the one that should be abandonded.
- */
+ * Stop the start up protocol. This method is called by the DataPath
+ * relaying a call from the VatTPMgr when it determines that
+ * there are two connections being built between this vat and another vat,
+ * andthat this connection is the one that should be abandonded.
+ */
/*package*/ void stopStartUpProtocol() {
myStop = true;
1.3 +10 -10 e/src/jsrc/net/vattp/data/Suspend.java
Index: Suspend.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/Suspend.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- Suspend.java 2001/09/06 09:55:39 1.2
+++ Suspend.java 2001/10/02 01:29:18 1.3
@@ -35,12 +35,12 @@
/*package*/ class Suspend implements MsgHandler {
/**
- * Constuct an object to handle the SUSPEND message (which also acts as
- * the shutdown message).
- *
- * @param connection is the DataConnection to work with.
- */
- /*package*/ Suspend(DataConnection connection) {
+ * Constuct an object to handle the SUSPEND message (which also acts as
+ * the shutdown message).
+ *
+ * @param connection is the VatTPConnection to work with.
+ */
+ /*package*/ Suspend(VatTPConnection connection) {
try {
connection.registerMsgHandler(Msg.SUSPEND, this);
} catch(IOException e) {
@@ -51,9 +51,9 @@
/**
* Handle a dead connection.
*
- * @param connection is the DataConnection which died.
+ * @param connection is the VatTPConnection which died.
*/
- public void connectionDead(DataConnection connection, Throwable reason) {
+ public void connectionDead(VatTPConnection connection, Throwable reason) {
// nothing to do
}
//Methods from the MsgHandler interface
@@ -62,9 +62,9 @@
* Handle incoming Suspend messages
*
* @param message is the incoming message.
- * @param connection is the DataConnection which died.
+ * @param connection is the VatTPConnection which died.
*/
- public void processMessage(byte[] message, DataConnection connection) {
+ public void processMessage(byte[] message, VatTPConnection connection) {
Assertion.test(Msg.SUSPEND == message[0], "Message not a SUSPEND message\n",
HexStringUtils.byteArrayToReadableHexString(message));
if (1 == message.length) { // Message is a shutdown message
1.3 +1 -1 e/src/jsrc/net/vattp/data/VatLocationLookup.java
Index: VatLocationLookup.java
===================================================================
RCS file: /cvs/e/src/jsrc/net/vattp/data/VatLocationLookup.java,v
retrieving revision 1.2
retrieving revision 1.3
diff -u -r1.2 -r1.3
--- VatLocationLookup.java 2001/09/06 09:55:39 1.2
+++ VatLocationLookup.java 2001/10/02 01:29:18 1.3
@@ -30,7 +30,7 @@
/**
* Get the locations for a Process Location Server lookup
* operation. This method will be called on an object which is
- * registered thru a ConnectionsMgr's setVatLocationLookup method
+ * registered thru a VatTPMgr's setVatLocationLookup method
* when the requested vatID is not the same as the vatID of the
* local vat.
*
1.1 e/src/jsrc/net/vattp/data/VatTPConnection.java
Index: VatTPConnection.java
===================================================================
package net.vattp.data;
/*
The contents of this file are subject to the Electric Communities E Open
Source Code License Version 1.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.communities.com/EL/.
Software distributed under the License is distributed on an "AS IS" basis,
WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for
the specific language governing rights and limitations under the License.
The Original Code is the Distributed E Language Implementation, released
July 20, 1998.
The Initial Developer of the Original Code is Electric Communities.
Copyright (C) 1998 Electric Communities. All Rights Reserved.
Contributor(s): ______________________________________.
*/
// Copyright 1998, Electric Communities, all rights reserved worldwide.
import net.vattp.security.ESecureRandom;
import org.erights.e.develop.assertion.Assertion;
import org.erights.e.develop.trace.Trace;
import org.erights.e.elib.prim.Runner;
import org.erights.e.elib.tables.ConstList;
import org.erights.e.elib.util.ConditionLock;
import org.erights.e.elib.util.DynamicCollection;
import org.erights.e.elib.util.DynamicCollectionEnumeration;
import org.erights.e.elib.util.HexStringUtils;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.ConnectException;
import java.net.NoRouteToHostException;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.security.KeyPair;
import java.util.Hashtable;
import java.util.Vector;
/**
* Allow sending and receiving data to a single remote vat.
*
* <p>The important method(s) are shutDownConnection(),
* sendMsg, and registerMsgHandler.
*
* <p>The valid message types are defined in class Msg.
* @author Bill Frantz
*/
public final class VatTPConnection {
/** The VatTPMgr we report status changes to. */
private VatTPMgr myConnMgr;
/** The DataPath object we are using. During crossed connection
* resolution, the outbound path. */
private /*nullOK*/ DataPath myDataPath;
/** During crossed connection resolution, the incoming DataPath.
* Otherwise null. */
private /*nullOK*/ DataPath myIncomingDataPath;
/** Array of MsgHandlers indexed by message type */
private MsgHandler[] myMsgHandlers = new MsgHandler[Msg.HIGH_MSG_TYPE+1];
/** An enumeration of the places to search for the remote vat */
private DynamicCollectionEnumeration mySiteSearch;
/** A Hashtable of the InetAddresses tried which failed due to host
* related problems. They are not to be tried again in this attempt
* to connect to the vat. The key is InetAddress,
* data is InetAddress. */
private Hashtable myAddressesTried;
/** The first place to look for the remote vat when resuming the
* connection. This value is "" when there is no better guesses
* than come from a VLS lookup. */
private String myFirstAddressToTry = "";
/** The places to search for the remote vat as a semicolon separated
* list. */
private String myFlattenedRemoteSearchPath;
/** The location the current DataPath are (to) (have been) speaking
to. */
private String myRemoteAddr;
/** The InetAddress of the remote end, or null if no connection has been
made */
private NetAddr myRemoteNetAddr = null;
/** The InetAddress of the local end, or null if no connection has been
made */
private NetAddr myLocalNetAddr = null;
/** The VatID of the other end of the connectiion. May be null. */
private String myRemoteVatID;
/** The runner to have the SendThread and RecvThread synchronize on
* when calling methods in this object.
*/
private Runner myRunner;
/** The key pair which defines the identity of this vat */
private KeyPair myIdentityKeys;
private String myLocalVatID;
/** The semicolon separated search path for this vat */
private String myLocalFlattenedSearchPath;
/** The reason shutDownConnection has been called */
private Throwable myShutdownReason = null;
static /*package*/ final byte STARTING = 1;
static /*package*/ final byte RUNNING = 2;
static /*package*/ final byte SUSPENDING = 3;
static /*package*/ final byte SUSPENDED = 4;
static /*package*/ final byte RESUMING = 5;
static /*package*/ final byte DIEING = 6;
static /*package*/ final byte DEAD = 7;
/** The current state of this VatTPConnection */
private byte myState;
/** Place to queue outgoing messages while myState != RUNNING */
private Vector myPendingOutput = null;
/** Place to accumulate the results of our search attempts */
private Vector myProblemAccumulator;
/** Whether we are initiating the connection or responding to a remotely
* initiated connection. */
private boolean myIsIncoming;
/** The suspend ID to present to the other end for a resume, or null for
* a new connection */
private byte[] /*nullOK*/ myOutgoingSuspendID;
/** The suspend ID we asked the other end to present to resume the
* connection, or null if the connection is not suspended. */
private byte[] /*nullOK*/ myLocalSuspendID;
// The following field is used by RecvThread and SendThread.
// When the connection shuts down, the IVs are updated
// so the connection can be resumed from suspension. The authentication
// keys, and protocol versions are set when the connection
// is first established.
/** The protocol versions and authorization parameters */
private AuthSecrets myProtocolParms;
// The following fields contain performance counters
/** The total bytes of messages received over this VatTPConnection */
/*package*/ long bytesReceived = 0;
/** The total number of messages received over this VatTPConnection */
/*package*/ long messagesReceived = 0;
/** The size of the largest message received over this VatTPConnection */
/*package*/ int maxReceivedMessageSize = 0;
/** The total bytes of messages sent over this VatTPConnection */
/*package*/ long bytesSent = 0;
/** The total number of messages sent over this VatTPConnection */
/*package*/ long messagesSent = 0;
/** The size of the largest message sent over this VatTPConnection */
/*package*/ int maxSentMessageSize = 0;
/**
* Make a new VatTPConnection
*
* @param connMgr The VatTPMgr to notify of status
* changes.
*
* @param remoteVatID the vatID of the remote vat
*
* @param searchPath an array of Strings each of which is the IP address
* and port number of a place to look for the remote vat. The IP address
* can be a DNS name or a dot repsentation of the 32 bit IP number. Most
* commonly, the Strings will be the location of the Vat Location
* Servers (VLSs) with which the remote vat is believed to register.
*
* @param identityKeys is the KeyPair which defines the identity of this
* vat.
*
* @param runner is the Runner whose thread we use to synchronize calls
* into the VatTPConnection object from the SendThread and ReceiveThread
*
* @param localFlattenedSearchPath Is the search path we publish for
* references to us, flattened into a string.
*
* @param localSearchPath Is the search path we publish for references
* to us, with each element a separate String in an array.
*/
/*package*/ VatTPConnection(VatTPMgr connMgr,
String remoteVatID,
String flattenedRemoteSearchPath,
KeyPair identityKeys,
String localVatID,
Runner runner,
String localFlattenedSearchPath)
{
myConnMgr = connMgr;
myRemoteVatID = remoteVatID;
myIsIncoming = false;
myIdentityKeys = identityKeys;
myLocalVatID = localVatID;
myRunner = runner;
myOutgoingSuspendID = null;
myLocalFlattenedSearchPath = localFlattenedSearchPath;
myState = STARTING;
//Save to resume the connection
myFlattenedRemoteSearchPath = flattenedRemoteSearchPath;
//Make a list of locations to try for the new connection
// myFirstAddressToTry will always be "" here.
DynamicCollection searchCollection
= new DynamicCollection(
EARL.parseSearchPath(myFlattenedRemoteSearchPath));
mySiteSearch = searchCollection.elems();
myAddressesTried = new Hashtable(1);
// Make a place to keep the search results
myProblemAccumulator = new Vector(1,1);
// Start the search for the remote ID.
tryNextAddress();
if (Trace.comm.usage && Trace.ON) {
Trace.comm.usagem("VatTPConnection constructor done " + this);
}
}
/**
* Make a new VatTPConnection for an incoming connection
*
* @param connMgr The VatTPMgr to notify of status
* changes.
*
* @param identityKeys is the KeyPair which defines the identity of this
* vat.
*
* @param localVatID is the vatID of the local vat.
*
* @param runner is the Runner whose thread we use to synchronize calls
* into the VatTPConnection object from the SendThread and ReceiveThread
*
* @param remoteVatID is the vatID of the remote vat.
*
* @param path The DataPath object controlling the TCP connection.
*
* @param localFlattenedSearchPath Is the search path we publish for
* references to us, flattened into a string.
*
* @param localSearchPath Is the search path we publish for references
* to us, with each element a separate String in an array.
*
* @param isIncoming is true if this connection is an incoming connection,
* otherwise false.
*/
/*package*/ VatTPConnection(VatTPMgr connMgr,
KeyPair identityKeys,
String localVatID,
Runner runner,
String remoteVatID,
DataPath path,
String localFlattenedSearchPath,
boolean isIncoming)
throws IOException {
myConnMgr = connMgr;
myIsIncoming = isIncoming;
myIdentityKeys = identityKeys;
myLocalVatID = localVatID;
myRunner = runner;
myRemoteVatID = remoteVatID;
myLocalFlattenedSearchPath = localFlattenedSearchPath;
myState = STARTING;
// Make a place to keep the search results
myProblemAccumulator = new Vector(1,1);
connectPath(path, remoteVatID, isIncoming);
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("VatTPConnection constructor done " + this);
}
}
/**
*
*/
public ConditionLock embargoLock() {
return myDataPath.embargoLock();
}
/**
* Close the TCP link on this connection.
*/
/*package*/ void close() throws IOException {
myDataPath.shutDownPath();
}
/**
* Connect a DataPath object for an incoming connection to an existing
* VatTPConnection object. This call is where we first know that there
* is a crossed connection problem.
*
* @param incomingPath is the DataPath object for the incoming connection.
* @param remoteVatID is the vatID of the remote vat.
* @param isIncoming is true if the DataPath represents an incoming TCP
* connection, is false if it is an outgoing TCP connection.
* @return LIVES_CONTINUE - Continue setting up this connection
* <br> LIVES_DUP - This connection is a duplicate, discard it.
* <br> LIVES_NOTIFY - Notify the other end of a duplicate connection.
* The other end must decide which connection to keep.
*/
/*package*/ int connectPath(DataPath incomingPath,
String remoteVatID, boolean isIncoming)
throws IOException {
int ret; // Our return value
Assertion.test(remoteVatID.equals(myRemoteVatID), "Wrong VatTPConnection\n "
+ this + "\n " + incomingPath + "\n " + remoteVatID);
if (null == myDataPath) {
myDataPath = incomingPath;
myFlattenedRemoteSearchPath
= myDataPath.connectConnection(this, myProtocolParms);
if (SUSPENDED == myState) myState = RESUMING;
return VatTPMgr.LIVES_CONTINUE; //New path is only one
}
// We must decide which connection to keep
int outState = myDataPath.getStartupState();
if (!isIncoming) {
ret = VatTPMgr.LIVES_DUP; //Zap anonymous outgoing
} else if (StartUpProtocol.ST_OUTGOING_EXPECT_GOTOO == outState) {
ret = VatTPMgr.LIVES_DUP; //Outgoing is too far along.
} else if (StartUpProtocol.ST_EXPECT_MESSAGE == outState) {
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("Killing incoming\n " + incomingPath
+ "\n in favor of\n " + myDataPath);
}
ret = VatTPMgr.LIVES_DUP; //Outgoing is too far along.
} else if (0 < myRemoteVatID.compareTo(myLocalVatID) ) {
// The far end is in charge
ret = VatTPMgr.LIVES_NOTIFY;
myIncomingDataPath = incomingPath;
myFlattenedRemoteSearchPath
= myIncomingDataPath.connectConnection(this, myProtocolParms);
} else {
// We must decide. Outbound not EXPECT_GOTO or EXPECT_MESSAGE,
// That means we keep the incoming and stop the outgoing.
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("Killing outgoing\n " + myDataPath
+ "\n in favor of\n " + incomingPath);
}
myDataPath.stopStartUpProtocol();
myDataPath = incomingPath;
myFlattenedRemoteSearchPath
= myDataPath.connectConnection(this, myProtocolParms);
ret = VatTPMgr.LIVES_CONTINUE;
}
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("Returning "+ ret + " for " + incomingPath);
}
return ret;
}
/**
* Duplicate crossed connection DataPath exiting.
*
* @param reason is the reason this is a duplicate.
* @param path is the path which has been declared duplicate.
*/
/*package*/ void duplicatePath(String reason, DataPath path) {
recordConnectionFailure(reason);
if (myDataPath == path) myDataPath = myIncomingDataPath;
myIncomingDataPath = null;
}
/**
* Enqueue a message for output.
*
* @param the message to queue
* @exception throws IOException if state is DEAD or DIEING
*/
private void enqueue(Object message) throws IOException {
if (DEAD == myState || DIEING == myState) {
throw new IOException("Dead or dying VatTPConnection, state=" + myState);
}
if (RUNNING == myState) {
int sus = myDataPath.getStartupState();
Assertion.test(StartUpProtocol.ST_EXPECT_MESSAGE == sus, "state =" + sus);
myDataPath.enqueue(message);
} else {
if (null == myPendingOutput) myPendingOutput = new Vector(5);
myPendingOutput.addElement(message);
if (SUSPENDED == myState) {
startResume();
}
}
}
//The following routines are used by StartUpProtocol to communicate is state
//success, failure etc., to make inquiries as to the state of other
//VatTPConnection objects, and to send messages.
/**
* Add an address to the list of addresses to try.
*
* @param addresses An array of Strings of the addresses to add.
*/
/*package*/ void extendSearchPath(ConstList addresses) {
mySiteSearch.addElems(addresses);
}
/**
* Get the NetAddr of the local end, or null if no connection has been
* made. If the connection is suspended or dead, this address will be
* the last address used when there was a live connection.
*
* @return the NetAddr of the local end or null if no connection has been
* made.
*/
public /*nullOK*/ NetAddr getLocalNetAddr() {
return myLocalNetAddr;
}
/**
* Get the vatID of the local end.
*
* @return the vatID of the local identity as a string.
*/
public String getLocalVatID() {
return myLocalVatID;
}
/**
* Get the outgoing suspend ID for resuming a connection.
*
* @return is the suspend ID or null if the connection is not suspended.
*/
/*package*/ byte[] /*nullOK*/ getOutgoingSuspendID() {
return myOutgoingSuspendID;
}
/**
* Get the NetAddr of the remote end, or null if no connection has been
* made. If the connection is suspended or dead, this address will be
* the last address used when there was a live connection.
*
* @return the NetAddr of the remote end or null if no connection has been
* made.
*/
public /*nullOK*/ NetAddr getRemoteNetAddr() {
return myRemoteNetAddr;
}
/**
* Get the search path to the remote end
*/
public ConstList getRemoteSearchPath() {
return EARL.parseSearchPath(myFlattenedRemoteSearchPath);
}
/**
* Get the vatID of the remote end.
*
* @return the vatID of the remote end as a string.
*/
public String getRemoteVatID() {
return myRemoteVatID;
}
/** Get the state of this VatTPConnection.
*
* @return is the current state (STARTING, RUNNING, DIEING etc.) of this
* VatTPConnection object.
*/
/*package*/ int getState() {
return myState;
}
/**
* Handle the death of the connection
*/
private void handleConnectionDeath(Throwable reason) {
myState = DEAD; // Tell referencers to allow garbage collection
myConnMgr.deathNotification(this);
for (int i=0; i<myMsgHandlers.length; i++) {
MsgHandler h = myMsgHandlers[i];
if (null != h) {
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm(this + " calls connectionDead in " + h);
}
h.connectionDead(this, reason);
}
}
}
/**
* Suspend this connection in response to a request from the other end.
*
* @param suspendID is the ID we will need to present to resume the connection.
*/
/*package*/ void handleSuspend(byte[] suspendID) throws IOException {
myOutgoingSuspendID = suspendID;
if (RUNNING == myState) {
suspend();
} else if (SUSPENDING != myState) {
Trace.comm.errorm("Invalid state for handleSuspend=" + myState);
}
myDataPath.shutDownPath();
myDataPath = null;
}
/**
* Notice that an identified incoming DataPath has failed to complete the
* startup protocol.
*
* @param is the DataPath object which failed.
*/
/*package*/ void identifiedPathDied(String reason, DataPath path) {
if (path == myIncomingDataPath) myIncomingDataPath = null;
if (path == myDataPath) {
}
}
/**
* Decide whether to keep incoming connection or outgoing when other end
* asks us to chose on an outgoing connection.
*
* @param vatID the vat ID of the remote vat.
* @return true - Keep the incoming connection of the pair.
* <br> false - Keep the outgoing connection of the pair.
*/
/*package*/ boolean isChoiceIncoming() throws IOException {
VatTPConnection incoming = null;
if (null == myIncomingDataPath) {
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("Returning false, no incoming");
}
return false;
} else {
int inState = myIncomingDataPath.getStartupState();
if (StartUpProtocol.ST_INCOMING_EXPECT_GO == inState) {
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("Returning true, incoming expecting go");
}
myDataPath.stopStartUpProtocol();
myDataPath = myIncomingDataPath;
myIncomingDataPath = null;
myDataPath.connectConnection(this, myProtocolParms);
return true; //Incoming is too far along.
} else if (StartUpProtocol.ST_EXPECT_MESSAGE == inState) {
Trace.comm.errorm("Incoming running, why are we chosing?"
+ " IncomingState=" + inState);
myDataPath.stopStartUpProtocol();
myDataPath = myIncomingDataPath;
myIncomingDataPath = null;
myDataPath.connectConnection(this, myProtocolParms);
return true;
} else {
myIncomingDataPath.stopStartUpProtocol();
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("Returning false, killing incoming");
}
myIncomingDataPath = null;
return false;
}
}
}
/*
* Make a report of the results of all the connection attempts
*
* @return a report of the connection results for each TCP connection,
* one connection attempt per line.
*/
private String makeConnectionStatusReport() {
if (null == myProblemAccumulator) {
return "**Running Connection**";
}
if (0 == myProblemAccumulator.size()) {
if (myIsIncoming) {
return "**Incoming Connection**";
} else {
return "**Outbound - No Status Reported";
}
}
StringBuffer connectStatusReport = new StringBuffer(300);
for (int i=0; i<myProblemAccumulator.size(); i++) {
Object problem = myProblemAccumulator.elementAt(i);
if (problem instanceof String) {
connectStatusReport.append(problem).append('\n');
} else {
Throwable evt = (Throwable) problem;
String subMsg = evt.getMessage();
if (evt instanceof NoRouteToHostException) ;
else if (evt instanceof UnknownHostException) ;
else if (evt instanceof ConnectException
&& (subMsg.startsWith("Connection refused"))) ;
else if (evt instanceof SocketException
&& (subMsg.startsWith("Connection reset by peer"))) ;
else {
Trace.comm.errorm("Error during connection attempt", evt);
}
connectStatusReport.append(evt.toString()).append('\n');
}
}
return connectStatusReport.toString();
}
/**
* Handle a new incoming message from the remote end.
*
* @param message the incoming message as a byte array. The first
* character is the message type as defined in class Msg.
*/
/*package*/ void newIncomingMsg(byte[] message) {
if (Trace.comm.verbose && Trace.ON) {
Trace.comm.verbosem(HexStringUtils.byteArrayToReadableHexString(message));
}
bytesReceived += message.length;
messagesReceived++;
if (maxReceivedMessageSize < message.length) {
maxReceivedMessageSize = message.length;
}
int msgType = message[0] & 0xff;
if (msgType <= Msg.HIGH_MSG_TYPE) {
MsgHandler handler = myMsgHandlers[msgType];
if (null != handler) {
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm(this + " calls processMessage in "
+ handler + "\n"
+ HexStringUtils.byteArrayToReadableHexString(message));
}
handler.processMessage(message, this);
return;
}
//Ignore unhandled messages during shutdown
if (DIEING == myState || DEAD == myState) return;
}
Trace.comm.errorm("No handler for incoming message type\n"
+ HexStringUtils.byteArrayToReadableHexString(message));
}
/**
* Notice problem on the connection. This method is called by the Send and
* Recv threads when they encounter a fatal error and give up.
*
* @param e The exception which describes the problem
*/
/*package*/ void noticeProblem(Throwable problem) {
switch (myState) {
case DEAD:
case DIEING:
case RUNNING:
myShutdownReason = problem;
break;
case RESUMING: // Same as STARTING
case STARTING:
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("accumulating problem report: "
+ this + ": ", problem);
}
recordConnectionFailure(problem);
break;
case SUSPENDING:
case SUSPENDED:
break;
default:
Trace.comm.errorm("Unhandled internal state=" + myState,
problem);
break;
}
}
/**
* Check if a connection should be resumed.
*
* @param suspendID The secret that the other side must know to resume a
* connection or null if this is a new connection.
* @return true if the connection should be completed.
* <p>false if the connection should be abandoned.
*/
/*package*/ boolean noticeRemoteResume(byte[] suspendID) {
if (null != suspendID) {
if (null == myLocalSuspendID) return false;
if (myLocalSuspendID.length != suspendID.length) return false;
for (int i=0; i<suspendID.length; i++) {
if (myLocalSuspendID[i] != suspendID[i]) return false;
}
return true;
}
if (null != myLocalSuspendID) return false;
return true;
}
/**
* Record why this try failed. When the path is shutdown, the next
* location in the search order will be tried.
*
* @param msg A String or Throwable describing the reason for
* abandoning this try.
*/
/*package*/ void recordConnectionFailure(Object msg) {
if (null != myProblemAccumulator) {
myProblemAccumulator.addElement(msg);
}
}
/**
* Register a MsgHandler for a particular message type. It will be called
* whenever a new message of the specified type arrives. It will also
* be called when the connection dies.
*
* @param msgType is the message type. It must be chosen from the types
* defined in the definition class Msg. An IOException is thrown if an
* attempt is made to register more
* than one handler for a message type.
*
* <p>An assertion is made that the msgType is not out of range.
* An assertion is made that no attempts are made to
* register a handler after the reactToNewConnection
* method of the object registered with the VatTPMgr has
* returned.
*
* @param handler is an object which implements the MsgHandler interface.
* It will be called with processMessage when a message
* of msgType is received from the remote vat. It will be called with
* connectionDead when this connection dies.
*
* @exception IOException is thrown if
* there is already a different handler registered for this message type.
*
* @see Msg
* @see MsgHandler
* @see VatTPMgr
*/
public void registerMsgHandler(byte msgType, MsgHandler handler)
throws IOException {
if (!myRunner.isCurrentThreadInVat()) {
Trace.comm.errorm("Caller doesn't hold vat lock", new Throwable());
}
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("registerMsgHandler="+ msgType +"("
+ handler +") on " + this);
}
Assertion.test(msgType > 0 || msgType <= Msg.HIGH_MSG_TYPE,
"msgType=" + msgType + " out of range (1 .. " + Msg.HIGH_MSG_TYPE
+ ")");
Assertion.test( STARTING == myState || RESUMING == myState,
"Called after the NewConnectionReactor returns\n handler="
+ handler + "\n myState=" + myState);
if (null != myMsgHandlers[msgType]
&& myMsgHandlers[msgType] != handler) {
throw new IOException(myMsgHandlers[msgType]
+ " already registered for msgType=" + msgType);
}
myMsgHandlers[msgType] = handler;
}
/**
* sendFinished called when a message has been completely passed to TCP
*
* @param length The size of the message passed to TCP.
* @param continuation The notification passed to sendMessage.
*/
/*package*/ void sendFinished(int count, int length,
StreamMessage /*NullOK*/ msg) {
bytesSent += length;
messagesSent += count;
if (maxSentMessageSize < length) maxSentMessageSize = length;
if (null != msg) {
msg.myPlaceToRun.enqueue(msg.myNotification);
}
}
/**
* Send a message to the remote vat.
*
* @param message is the message to be sent. It must not be altered after
* the call to sendMsg. The first byte of message is the message type and
* must be chosen from the types defined in the definition class Msg.
* (These two restrictions allow sendMsg to avoid copying the message
* into a private buffer.)
*
* @exception IOException is thrown if the message can not be queued to
* be sent because the connection has broken, if the message is longer
* than, Msg.MAX_OUTBOUND_MSG_LENGTH, or if there is no listener
* registered to process messages of this message's message type.
*
* @see Msg
*/
public void sendMsg(byte[] message) throws IOException {
if (!myRunner.isCurrentThreadInVat()) {
Trace.comm.errorm("Caller doesn't hold vat lock", new Throwable());
}
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("sendMsg on " + this + "\n"
+ HexStringUtils.byteArrayToReadableHexString(message));
}
if (message.length > Msg.MAX_OUTBOUND_MSG_LENGTH) {
throw new IOException("Outbound message length=" + message.length
+ " greater than limit="
+ Msg.MAX_OUTBOUND_MSG_LENGTH);
}
if (Trace.comm.debug && Trace.ON) Trace.comm.debugm("enqueueing to "
+ myRemoteAddr
+ "\n" + HexStringUtils.byteArrayToReadableHexString(message));
int msgType = message[0] & 0xff;
if (msgType > Msg.HIGH_MSG_TYPE || null == myMsgHandlers[msgType]) {
throw new IOException("No MsgHandler registered for message type="
+ msgType);
}
enqueue(message);
}
/**
* Send a message to the remote vat with notification that it has been
* queued for processing by the TCP/IP stack on the local machine.
* This call is designed to allow messaging with flow control. By only
* sending short message segments, and waiting for notification
* before sending the next segment,
* a caller can limit the buffering required by the communication system.
*
* @param message is the message to be sent. It must not be altered after
* the call to sendMsg. The first byte of message is the message type and
* must be chosen from the types defined in the definition class Msg.
* (These two restrictions allow sendMsg to avoid copying the message
* into a private buffer.)
* @param notification is the Runnable which will be enqueued when
* the message has been passed to the platform's TCP stack.
* @param placeToRun is the Runner where the notification will be
* enqueued.
*
* @exception IOException is thrown if the message can not be queued to
* be sent because the connection has broken, if the message is longer
* than, Msg.MAX_OUTBOUND_MSG_LENGTH, or if there is no listener
* registered to process messages of this message's message type.
*
* @see Msg
*/
public void sendMsg(byte[] message,
Runnable notification,
Runner placeToRun) throws IOException {
if (!myRunner.isCurrentThreadInVat()) {
Trace.comm.errorm("Caller doesn't hold vat lock", new Throwable());
}
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("sendMsg, Notify=" + notification + " Runner="
+ placeToRun + " on " + this + "\n"
+ HexStringUtils.byteArrayToReadableHexString(message));
}
if (message.length > Msg.MAX_OUTBOUND_MSG_LENGTH) {
throw new IOException("Outbound message length=" + message.length
+ " greater than limit="
+ Msg.MAX_OUTBOUND_MSG_LENGTH);
}
if (Trace.comm.debug && Trace.ON) Trace.comm.debugm("enqueueing to "
+ myRemoteAddr
+ "\n" + HexStringUtils.byteArrayToReadableHexString(message));
int msgType = message[0] & 0xff;
if (msgType > Msg.HIGH_MSG_TYPE || null == myMsgHandlers[msgType]) {
throw new IOException("No MsgHandler registered for message type="
+ msgType);
}
enqueue(new StreamMessage(message, notification, placeToRun));
}
/**
* Causes a clean shutdown of the connection. All messages sent
* (XXX markm asks: "All messages sent" or "All messages queued to be
* sent"?)
* previously will be sent. When the shut down is complete, the
* connectionDead method will be called on all registered MsgHandlers.
*/
public void shutDownConnection() {
shutDownConnection(new ConnectionShutDownException(
"This end requested shutdown"));
}
/**
* Causes a clean shutdown of the connection. All messages sent
* (XXX markm asks: "All messages sent" or "All messages queued to be
* sent"?)
* previously will be sent. When the shut down is complete, the
* connectionDead method will be called on all registered MsgHandlers.
* <p>
* XXX markm: I made this public, and changed ProxyConnection to call
* this with a non-ConnectionShutDownException. As far as I can tell,
* this shouldn't break anything. But Bill, it would be good for you
* to double check this.
*
* @param reason indicates why the connection is shutting down
*/
public void shutDownConnection(Throwable reason) {
if (!myRunner.isCurrentThreadInVat()) {
Trace.comm.errorm("Caller doesn't hold vat lock", new Throwable());
}
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("shutDownConnection " + this);
}
Assertion.test(null != reason, "Must provide a reason for shutdown");
if (RUNNING == myState) {
myDataPath.enqueue(new byte[] {Msg.SUSPEND});
myState = DIEING;
myConnMgr.enterHospice(this, myRemoteVatID);
myShutdownReason = reason;
} else if (SUSPENDED == myState) {
startResume(); // Resume it to kill it
myShutdownReason = reason;
}
}
/**
* shutDownFinished called when all queued messages have been sent and
* the socket closed.
*/
/*package*/ void shutDownFinished(Throwable reason) {
if (STARTING == myState) {
if (!myIsIncoming) {
tryNextAddress();
}
} else if (SUSPENDING == myState) {
myConnMgr.connectionSuspended(this, myRemoteVatID);
myState = SUSPENDED;
if (Trace.comm.usage && Trace.ON) {
Trace.comm.usagem("connection suspended " + this);
}
if (null != myPendingOutput) { // If we have pending output
startResume(); // try to resume
}
} else {
Throwable rr = (reason instanceof ConnectionShutDownException
&& null != myShutdownReason)
? myShutdownReason
: reason;
if (Trace.comm.usage && Trace.ON) {
if (Trace.comm.event) {
Trace.comm.eventm(
"Connection died " + this
+ "\n" + myConnMgr.toString()
+ "\n" + rr);
} else {
Trace.comm.usagem(
"Connection died " + myRemoteAddr + "|"
+ myRemoteVatID + "\n" + rr);
}
}
handleConnectionDeath(rr);
}
}
/**
* Start trying to resume the connection.
*/
private void startResume() {
if (Trace.comm.usage && Trace.ON) {
Trace.comm.usagem("Resuming connection " + this);
}
myIsIncoming = false;
myState = RESUMING;
//Make a list of locations to try for the new connection
String path = (myFirstAddressToTry.equals("")
? myFlattenedRemoteSearchPath
: myFirstAddressToTry + ";" + myFlattenedRemoteSearchPath);
DynamicCollection searchCollection
= new DynamicCollection(EARL.parseSearchPath(path));
mySiteSearch = searchCollection.elems();
myAddressesTried = new Hashtable(1);
// Make a place to keep the search results
myProblemAccumulator = new Vector(1,1);
// Start the search for the remote ID.
tryNextAddress();
}
/**
* Called by StartUpProtocol when a connection to a remote vat has
* been successfully made. The StartUpProtocol object has unregistered
* its MsgHandlers and is ready to become garbage.
*
* @param dataPath is the DataPath object which made the connection.
* @param remoteVatID is the vatID of the remote vat.
* @param remoteNetAddr is the remote end's network address.
* @param locaNetAddr is the local end's network address.
* @param remotePort is the remote port number we tried to connect to,
* or 0 if this is an inbound connection (or the port number
* is unavailable for some other reason). This port number
* will be used if we need to resume the connection later.
* @param authParms is the AuthSecrets object which contains the
* protocol versions and authorization parameters for the connection.
*/
/*package*/ void startupSuccessful(
DataPath path,
String remoteVatID,
NetAddr remoteNetAddr,
NetAddr localNetAddr,
int remotePort,
AuthSecrets authParms)
{
Assertion.test(myRemoteVatID.equals(remoteVatID),
"DataPath's remote vatID=", remoteVatID,
" not equal my remote vatID=", myRemoteVatID);
myProtocolParms = authParms;
//it's safe to not check optInetAddress() for null, since
//remoteNetAddr must explicitly have one
myRemoteAddr = remoteNetAddr.optInetAddress().getHostAddress() + ":"
+ remoteNetAddr.getPort();
myRemoteNetAddr = remoteNetAddr;
myLocalNetAddr = localNetAddr;
if (null != myIncomingDataPath) {
// We have two DataPaths
if (path == myIncomingDataPath) {
myDataPath.stopStartUpProtocol();
myDataPath = myIncomingDataPath;
} else {
Assertion.test(path == myDataPath, path.toString(), "\n",
myDataPath.toString());
myIncomingDataPath.stopStartUpProtocol();
}
myIncomingDataPath = null;
}
if (remotePort > 0) { // We know where to retry to resume later
//it's safe to not check optInetAddress() for null, since
//remoteNetAddr must explicitly have one
myFirstAddressToTry
= remoteNetAddr.optInetAddress().getHostAddress()
+ ":" + remotePort;
}
if (RESUMING == myState) {
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("\nResume successfull after:\n"
+ makeConnectionStatusReport());
}
} else {
new Suspend(this); // Object to handle SUSPEND messages
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("\nStartup successfull after:\n"
+ makeConnectionStatusReport());
}
}
myConnMgr.startupSuccessful(this,
myRemoteVatID, RESUMING == myState);
myState = RUNNING;
if (null != myShutdownReason) { // Someone has asked to shut down
shutDownConnection(myShutdownReason);
return;
}
myProblemAccumulator = null;
if (null != myPendingOutput) {
for (int i=0; i<myPendingOutput.size(); i++) {
myDataPath.enqueue(myPendingOutput.elementAt(i));
}
myPendingOutput = null;
}
}
/**
* Suspend the connection. Note that this method is NOT part of the
* published API. It is public to enable testing of the
* suspend/resume logic. (Popular demand could cause it to remain
* public.)
*
* <p>Suspended connections will be automatically resumed when data
* is sent to them with sendMsg().
*
* @throws IOException if the connection isn't in the RUNNING state.
*/
public void suspend() throws IOException {
if (!myRunner.isCurrentThreadInVat()) {
Trace.comm.errorm("Caller doesn't hold vat lock", new Throwable());
}
if (RUNNING != myState) {
throw new IOException(
"Attempt to suspend a non-running connection, state="
+ myState);
}
myLocalSuspendID = new byte[20];
//MSM: should be passed in entropy as a capability
ESecureRandom entropy = ESecureRandom.getESecureRandom();
entropy.nextBytes(myLocalSuspendID);
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("Suspend Connection " + this
+ HexStringUtils.byteArrayToReadableHexString(
myLocalSuspendID));
}
ByteArrayOutputStream baos = new ByteArrayOutputStream();
DataOutputStream os = new DataOutputStream(baos);
os.writeByte(Msg.SUSPEND);
os.writeShort(myLocalSuspendID.length);
os.write(myLocalSuspendID);
os.flush();
enqueue(baos.toByteArray());
myConnMgr.connectionSuspending(this, myRemoteVatID);
myState = SUSPENDING;
}
/**
* Describe this connection object.
*
*/
public String toString() {
return super.toString() + "\n to " + myRemoteAddr + "|"
+ myRemoteVatID
+ "\n searchpath=" + myFlattenedRemoteSearchPath
+ " state=" + myState;
}
/**
* Start an attempt to connect to the next (first) address in the list
* of addresses to try.
*/
private void tryNextAddress () {
if (null != mySiteSearch && mySiteSearch.hasMoreElements()) {
myRemoteAddr = (String) mySiteSearch.nextElement();
//Build a DataPath to try the connection.
myDataPath = new DataPath(null, // No VatTPMgr.
this,
myRemoteVatID,
myRemoteAddr,
myAddressesTried,
myIdentityKeys,
myLocalVatID,
myRunner,
myOutgoingSuspendID,
myProtocolParms,
myLocalFlattenedSearchPath);
} else {
String report = makeConnectionStatusReport();
if (Trace.comm.usage && Trace.ON) {
Trace.comm.usagem(toString()
+ "\nConnection attempt failed, search path exhausted\n"
+ report);
}
handleConnectionDeath(new IOException(report));
}
}
/**
* Remove the registration for a handler for a particular message type.
*
* @param msgType is the message type. It must be chosen from the types
* defined in the definition class Msg.
* It is a fatal error if the msgType is out of range.
*
* @param handler is the object currently registered as the handler.
*
* @exception IOException is thrown if the handler passed is not the
* handler registered for this message type.
*
* @see Msg
* @see MsgHandler
*/
/*package*/ void unRegisterMsgHandler (byte msgType, MsgHandler handler)
throws IOException {
if (!myRunner.isCurrentThreadInVat()) {
Trace.comm.errorm("Caller doesn't hold vat lock", new Throwable());
}
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("unRegisterMsgHandler="+ msgType +"("
+ handler +") on " + this);
}
if (msgType <= 0 || msgType > Msg.HIGH_MSG_TYPE) {
Trace.comm.errorm("msgType=" + msgType +" out of range (1 .. "
+ Msg.HIGH_MSG_TYPE);
Trace.comm.notifyFatal();
}
if (handler != myMsgHandlers[msgType]) {
throw new IOException("Registered=" + myMsgHandlers[msgType]
+ " is not the same as " + handler + " for msgType=" + msgType);
}
myMsgHandlers[msgType] = null;
}
}
1.1 e/src/jsrc/net/vattp/data/VatTPMgr.java
Index: VatTPMgr.java
===================================================================
package net.vattp.data;
/*
The contents of this file are subject to the Electric Communities E Open
Source Code License Version 1.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.communities.com/EL/.
Software distributed under the License is distributed on an "AS IS" basis,
WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License for
the specific language governing rights and limitations under the License.
The Original Code is the Distributed E Language Implementation, released
July 20, 1998.
The Initial Developer of the Original Code is Electric Communities.
Copyright (C) 1998 Electric Communities. All Rights Reserved.
Contributor(s): ______________________________________.
*/
// Copyright 1998, Electric Communities, all rights reserved worldwide.
import org.erights.e.develop.assertion.Assertion;
import org.erights.e.develop.trace.Trace;
import org.erights.e.elib.prim.E;
import org.erights.e.elib.prim.Runner;
import org.erights.e.elib.ref.Ref;
import org.erights.e.elib.ref.Resolver;
import org.erights.e.elib.tables.ConstList;
import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.security.KeyPair;
import java.util.Enumeration;
import java.util.Hashtable;
/**
* Manage the connections between this vat and other vats.
*
* <p> The VatTPMgr acts as the focal point for managing the
* TCP connections from a particular vat. It uses the public/private
* key pair for the vat, which defines the vat's identity.
*
* <p>The important method(s) are getConnection(...) which returns a
* VatTPConnection to a remote vat, and addNewConnectionReactor
* which lets the higher layers connect themselves to connections
* which have completed the startup protocol.
*
* @author Bill Frantz
* @author Mark S. Miller
*/
public class VatTPMgr {
// Identity stuff
private KeyPair myIdentityKeys;
/** The vatID of the local vat. The hash of this end's public key.*/
private String myLocalVatID;
// VLS registration stuff
private NetConfig myNetConfig;
/** VLS registrations that we maintain */
private VatLocationLookup myVLS;
/** Connection to the E run queue */
private Runner myRunner;
/** The thread which is listening for incoming connections */
private ListenThread myListenThread;
/** The object to notify about newly available connections */
private NewConnectionReactor myReactor;
// The following fields maintain knowledge of the existing
// VatTPConnection objects. A VatTPConnection object start in either
// the myPendingOutgoingConnections or myPendingIncomingConnections
// table. When it finishes the startup protocol, it moves to the
// myRunningDataConnections table. When it start to shutdown, it moves
// to the myDieingConnections table. When it completes it's shutdown,
// it is removed from all tables so it may be garbage collected.
/** The unidentified connections which haven't completed the
* start up protocol. The key is the DataPath object. The value is
* the DataPath object if this entry is for an incoming connections,
* or the Resolver if the path was created for a connectToVatAt
* connection. */
private Hashtable /*nullOK*/ myUnidentifiedConnections;
/** The connections which haven't completed the start up protocol.
* The key is the vatID. The value is the VatTPConnection object. */
private Hashtable /*nullOK*/ myIdentifiedConnections;
/** The running DataConnections. The key is vatID, the value
* is the VatTPConnection object. */
private Hashtable myRunningDataConnections = new Hashtable(5);
/** The connections in the process of suspending. The key is vatID,
* the value is the VatTPConnection object. */
private Hashtable /*nullOK*/ mySuspendingConnections;
/** The connections which are suspended. The key is vatID,
* the value is the VatTPConnection object. */
private Hashtable /*nullOK*/ mySuspendedConnections;
/** Connections which have been told to shut down (until they have gone).
* The key is the VatTPConnection object and the value is the vatID.*/
private Hashtable /*nullOK*/ myDieingConnections;
static /*package*/ final int LIVES_CONTINUE = 1;
static /*package*/ final int LIVES_DUP = 2;
static /*package*/ final int LIVES_NOTIFY = 3;
/*
* Make a VatTPMgr listening on a specified port.
*
* Each VatTPMgr made will have a different ID. You only want
* to make one per vat.
*
* <p>The VatTPMgr is associated with one vat. It will ensure
* that its in-vat objects are synchronized with that vat's thread.
* This constructor gets the Runner for that vat via the static method
* Runner.currentRunner(). The rest of the VatTP system gets
* access to that runner by having the reference explicitly passed
* during construction. Therefore, if it is necessary to have a
* VatTP system work with a different runner, the constructor
* which accepts the Runner as a specific parameter should be used.
*
* @param identityKeys is the KeyPair which defines the identity of this
* vat.
* @param netConfig holds the configuration parameters for configuring
* how we interact with the network.
*/
public VatTPMgr(KeyPair identityKeys, NetConfig netConfig)
throws UnknownHostException, IOException {
this(identityKeys, netConfig, Runner.currentRunner());
}
/*
* Make a VatTPMgr.
*
* Each VatTPMgr made will have a different ID. You only want
* to make one per vat.
*
* <p>The VatTPMgr is associated with one vat. It will ensure
* that its in-vat objects are synchronized with that vat's thread.
*
* @param identityKeys is the KeyPair which defines the identity of this
* vat.
* @param netConfig holds the configuration parameters for configuring
* how we interact with the network.
* @param runner is the Runner to use to synchronize the connections.
*/
/*package*/ VatTPMgr(KeyPair identityKeys,
NetConfig netConfig,
Runner runner)
throws UnknownHostException, IOException {
myIdentityKeys = identityKeys;
myVLS = null;
myLocalVatID = VatIdentity.calculateVatID(identityKeys.getPublic());
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("netConfig is " + E.toString(netConfig));
}
myRunner = runner;
myListenThread = new ListenThread(netConfig.getOptListenAddr(),
this,
myRunner);
NetAddr addr = myListenThread.listenAddress();
myNetConfig = netConfig.withListenAddr(addr.toString(),
listenPath(addr));
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("VatTPMgr constructor done " + this);
}
}
/**
* Return the subset of a searchPath for directly connecting
*/
static private ConstList listenPath(NetAddr addr)
throws UnknownHostException
{
InetAddress[] ips = null;
InetAddress optIP = addr.optInetAddress();
if (optIP == null) {
InetAddress localhost = null;
try {
localhost = InetAddress.getLocalHost();
} catch (UnknownHostException uhe) {
//the loopback host: 127.0.0.1
localhost = InetAddress.getByName(null);
}
ips = InetAddress.getAllByName(localhost.getHostName());
} else {
InetAddress[] single = { optIP };
ips = single;
}
String suffix = ":" + addr.getPort();
String[] parts = new String[ips.length];
for (int i = 0; i < ips.length; i++) {
parts[i] = ips[i].getHostAddress() + suffix;
}
return ConstList.fromArray(parts);
}
/**
* Register that this connection is suspended
*
* @param conn The data connection which is suspended.
* @param vatID The vatID of the remote end.
*/
/*package*/ void connectionSuspended(VatTPConnection conn, String vatID) {
VatTPConnection sus = removeSuspending(vatID);
if (null == sus) {
Trace.comm.errorm("Suspended connection not suspending " + conn
+ "\n " + this, new Throwable("locator"));
} else if (sus != conn) {
Trace.comm.errorm("Suspending ID " + sus
+ " not one suspended " + conn);
}
if (null == mySuspendedConnections) {
mySuspendedConnections = new Hashtable(1);
}
mySuspendedConnections.put(vatID, conn);
}
/**
* Register that this connection is suspending
*
* @param conn The data connection which is suspending.
* @param vatID The vatID of the remote end.
*/
/*package*/ void connectionSuspending(VatTPConnection conn,
String vatID) {
VatTPConnection run
= (VatTPConnection)myRunningDataConnections.remove(vatID);
if (null == run) {
Trace.comm.errorm("Suspending connection not running " + conn
+ "\n " + this, new Throwable("locator"));
} else if (run != conn) {
Trace.comm.errorm("Running ID " + run
+ " not one suspending " + conn);
}
if (null == mySuspendingConnections) {
mySuspendingConnections = new Hashtable(1);
}
mySuspendingConnections.put(vatID, conn);
}
/**
* Notify the connectToVatAt caller that he has tried to connect to self.
*
* @param path is the DataPath object trying the connection.
*/
/*package*/void connectToSelf(DataPath path) {
// Remove from unidentified connections list
Resolver res = (Resolver)removeUnidentified(path);
if (null == res) {
Trace.comm.errorm("DataPath not known" + path);
} else {
res.resolve(null);
}
}
/**
* Make an authenticated connection to a vat given an IP:port address.
*
* @param ipPort is the IP address and port to connect to separated by a
* colon. For example "groucho.communities.com:1670".
* @return A promise for the resulting VatTPConnection
*/
public Ref connectToVatAt(String ipPort) {
Object[] promise = Ref.promise();
Ref result = (Ref)promise[0];
Resolver resolver = (Resolver)promise[1];
connectToVatAt(ipPort, resolver);
return result;
}
/**
* Make an authenticated connection to a vat given an IP:port address.
*
* @param ipPort The IP address and port to connect to, separated by a
* colon. For example "groucho.communities.com:1670".
* @param res The object that will be notified when the connection is
* completed or if the attempt to connect fails.
*/
public void connectToVatAt(String ipPort, Resolver res) {
if (!myRunner.isCurrentThreadInVat()) {
Trace.comm.errorm("Caller doesn't hold vat lock", new Throwable());
}
DataPath path = new DataPath(this,
null,
VatIdentity.WHOEVER,
ipPort,
new Hashtable(1),
myIdentityKeys,
myLocalVatID,
myRunner,
null,
null,
EARL.flattenSearchPath(searchPath()));
// Register the DataPath as an unidentified path
if (null == myUnidentifiedConnections) {
myUnidentifiedConnections = new Hashtable(1);
}
myUnidentifiedConnections.put(path, res);
}
/**
* Mark the death of a connection. Remove it from all tables so it can be
* garbage collected.
*
* @param conn The data connection which has shut down.
*/
/*package*/ void deathNotification(VatTPConnection conn) {
Object te = removeDieing(conn); //To find which table we
//removed if from.
if (null != te) return; // Tables cleaned up
String vID = conn.getRemoteVatID();
te = myRunningDataConnections.remove(vID);
if (null != te) {
if (te != conn) {
Trace.comm.errorm("Two DataConnections for one vatID\nA="
+ conn + "\nB=" + te);
}
return;
}
te = removeIdentified(vID);
if (null != te) {
if (te != conn) {
Trace.comm.errorm(
"Two DataConnections for one vatID\nA="
+ conn + "\nB=" + te);
}
return;
}
te = removeSuspending(vID);
if (null != te) {
if (te != conn) {
Trace.comm.errorm(
"Two DataConnections for one vatID\nA="
+ conn + "\nB=" + te);
}
return;
}
te = removeSuspended(vID);
if (null != te) {
if (te != conn) {
Trace.comm.errorm(
"Two DataConnections for one vatID\nA="
+ conn + "\nB=" + te);
}
return;
}
Trace.comm.errorm("Unregistered VatTPConnection=" + conn
+ "\n" + toString(), new Throwable("locator"));
}
/**
* Register that this connection is dying
*
* @param conn The data connection which is shutting down.
* @param vatID The vatID of the remote end.
*/
/*package*/ void enterHospice(VatTPConnection conn, String vatID) {
if (null == myDieingConnections) {
myDieingConnections = new Hashtable(1);
}
myDieingConnections.put(conn, vatID);
myRunningDataConnections.remove(vatID);
removeSuspending(vatID);
removeSuspended(vatID);
}
/**
* Find an existing VatTPConnection object that can or will be able to
* be used for data transfer. (That is, one that is not dying or dead.)
*
* @param vatID is the vatID of the remote vat.
* @return an existing VatTPConnection object or null if there are none.
*/
private /*nullOK*/ VatTPConnection findDataConnection(String vatID) {
VatTPConnection ret;
ret = (VatTPConnection)myRunningDataConnections.get(vatID);
if (null != ret) return ret;
if (null != mySuspendingConnections) {
ret = (VatTPConnection)mySuspendingConnections.get(vatID);
if (null != ret) return ret;
}
if (null != mySuspendedConnections) {
ret = (VatTPConnection)mySuspendedConnections.get(vatID);
if (null != ret) return ret;
}
if (null != myIdentifiedConnections) {
ret = (VatTPConnection)myIdentifiedConnections.get(vatID);
}
return ret;
}
/**
* Return a VatTPConnection to the remote vat given its vatID and
* search path.
*
* @param vatID is the vatID for the remote vat wanted. The
* process of setting up the physical connection will verify that the
* vatID is indeed the hash of the public key of the remote vat,
* and that the other end of the connection holds the associated private
* key.
*
* @param searchList is a list of Strings each of which is the IP
* address and port number of a place to look for the remote vat.
* The IP address can be a DNS name or a dot repsentation of the
* 32 bit IP number. Most commonly, the Strings will be the
* location of the Vat Location Servers (VLSs) with which the
* remote vat is believed to register. It may be useful under
* special circumstances to include the actual address where the
* remote vat may be listening in the list.
*
* @return a VatTPConnection object to the remote vat or null. This is
* either an existing VatTPConnection object or a newly created one. If
* the VatTPConnection object is newly created, then it has been asked to
* initiate the connection startup protocol. This method returns null if
* the request is to connect the vat to itself.
*/
public VatTPConnection getConnection(String vatID, ConstList searchList) {
if (!myRunner.isCurrentThreadInVat()) {
Trace.comm.errorm("Caller doesn't hold vat lock", new Throwable());
}
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("getConnection to=" + vatID
+ " on " + this);
}
VatTPConnection dc = findDataConnection(vatID);
if (null != dc) {
return dc;
}
if (vatID.equals(myLocalVatID)) {
return null;
}
dc = new VatTPConnection(this,
vatID,
EARL.flattenSearchPath(searchList),
myIdentityKeys,
myLocalVatID,
myRunner,
EARL.flattenSearchPath(searchPath()));
if (null == myIdentifiedConnections) {
myIdentifiedConnections = new Hashtable(1);
}
myIdentifiedConnections.put(vatID, dc);
return dc;
}
/**
* Like getConnection((vatID, searchList), but if there isn't a working
* live one, will return null rather than creating one.
*/
public VatTPConnection optConnection(String vatID) {
if (!myRunner.isCurrentThreadInVat()) {
Trace.comm.errorm("Caller doesn't hold vat lock", new Throwable());
}
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("optConnection to=" + vatID
+ " on " + this);
}
return findDataConnection(vatID);
}
/**
* List the current DataPaths/DataConnections in a enumeration.
*
* @param enum is the Enumeration to list. The elements produced should
* be either DataConnections or DataPaths.
* @param buf is a StringBuffer to fill with the list.
*/
private void listState(Enumeration enum, StringBuffer buf) {
while (enum.hasMoreElements()) {
Object element = enum.nextElement();
buf.append(element.toString()).append("\n");
}
}
/**
* Move a path from the unidentified list to the identified but not
* connected path list. This also handles forwarding to the promise
* returned by connectToVatAt for that type of connection.
*
* @param path is the DataPath object which has been identified.
* @param remoteVatID is the identified remote vat's vatID.
* @param isIncoming says whether this is an incoming connection (true)
* or an outgoing connection (false).
* @return LIVES_CONTINUE - Continue setting up this connection
* <br> LIVES_DUP - This connection is a duplicate, discard it.
* <br> LIVES_NOTIFY - Notify the other end of a duplicate connection.
* The other end must decide which
* connection to keep.
*/
private int moveIdentifiedPath(DataPath path,
String remoteVatID,
boolean isIncoming)
throws IOException {
int ret;
// Remove from unidentified connections list
Object uidPath = removeUnidentified(path);
if (null == uidPath) {
Trace.comm.errorm("DataPath not known" + path);
}
VatTPConnection dc = findDataConnection(remoteVatID);
if (null == dc) {
dc = new VatTPConnection(this,
myIdentityKeys,
myLocalVatID,
myRunner,
remoteVatID,
path,
EARL.flattenSearchPath(searchPath()),
isIncoming);
Trace.comm.debugm("New VatTPConnection " + dc + " for incoming="
+ isIncoming + " " + path);
// Insert the VatTPConnection in pending connections
if (null == myIdentifiedConnections) {
myIdentifiedConnections = new Hashtable(1);
}
myIdentifiedConnections.put(remoteVatID, dc);
ret = LIVES_CONTINUE;
} else {
ret = dc.connectPath(path, remoteVatID, isIncoming);
}
if (uidPath instanceof Resolver) {
Resolver res = (Resolver)uidPath;
res.resolve(dc);
}
return ret;
}
/**
* Register a DataPath for an connection where the remote vat has been
* identified
*
* @param path the new DataPath
* @param remoteVatID is the vat ID of the remote vat.
* @param localVatID is the vat ID of this vat.
* @param isIncoming is true if this is an inbound connection, false if
* it is an outbound connection.
*
* @return LIVES_CONTINUE - Continue setting up this connection
* <br> LIVES_DUP - This connection is a duplicate, discard it.
* <br> LIVES_NOTIFY - Notify the other end of a duplicate connection.
* The other end must decide which
* connection to keep.
*/
/*package*/ int newConnectionIdentified(DataPath path,
String remoteVatID,
String localVatID,
boolean isIncoming)
throws IOException {
if (Trace.comm.debug && Trace.ON) {
Trace.comm.debugm("(" + path + ") on " + this);
}
return moveIdentifiedPath(path, remoteVatID, isIncoming);
}
/**
* Receive a new inbound socket from the ListenThread.
*
* @param socket is the new inbound socket object, just as received by
* the listen operation.
*/
/*package*/ void newInboundSocket(Socket socket) {
DataPath path = new DataPath(this,
socket,
myIdentityKeys,
myLocalVatID,
myRunner,
EARL.flattenSearchPath(searchPath()),
myVLS);
if (null == myUnidentifiedConnections) {
myUnidentifiedConnections = new Hashtable(1);
}
myUnidentifiedConnections.put(path, path);
}
/**
* Learn of a problem with the ListenThread.
*
* @param error the Throwable which represents the error.
*/
/*package*/ void noticeProblem(Throwable error) {
//I (wsf) don't think there is anything to do here
Trace.comm.errorm("Fatal error reported", error);
Trace.comm.notifyFatal();
}
/**
* Register an object to notice a new connection which has
* successfully completed the startup protocol. This object is
* expected to connect up MsgHandlers for all the messages
* processed by higher layers in the protocol. Note that a
* maximum of one NewConnectionReactor may be registered. <p>
*
* This notification is for all DataConnections, whether the
* connection was initiated from this end or from the other end.
*
* @param reactor is an object which implements
* NewConnectionReactor.
* @exception IOException is thrown if there is already a reactor
* registered.
*/
public void addNewConnectionReactor(NewConnectionReactor reactor)
throws IOException {
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("addNewConnectionReactor " + reactor);
}
if (null != myReactor) {
throw new IOException("Must only addNewConnectionReactor() once");
}
myReactor = reactor;
}
/**
* Remove an existing VatTPConnection object that can or will be able to
* be used for data transfer (That is, one that is not dying or dead)
* from the hash table which holds it and return it.
*
* @param vatID is the vatID of the remote vat.
* @return an existing VatTPConnection object or null if there are none.
*/
private /*nullOK*/ VatTPConnection removeDataConnection(String vatID) {
VatTPConnection ret;
ret = (VatTPConnection)myRunningDataConnections.remove(vatID);
if (null != ret) return ret;
ret = removeSuspending(vatID);
if (null != ret) return ret;
ret = removeSuspended(vatID);
if (null != ret) return ret;
ret = removeIdentified(vatID);
return ret;
}
/** Remove an entry from myDieingConnections
*
* @param regID is the VatID to remove.
*/
private String /*nullOK*/ removeDieing(VatTPConnection dc) {
if (null == myDieingConnections) return null;
String ret
= (String)myDieingConnections.remove(dc);
if (0 == myDieingConnections.size()) {
myDieingConnections = null;
}
return ret;
}
/** Remove an entry from myIdentifiedConnections
*
* @param regID is the VatID to remove.
*/
private VatTPConnection /*nullOK*/ removeIdentified(String regID) {
if (null == myIdentifiedConnections) return null;
VatTPConnection ret
= (VatTPConnection)myIdentifiedConnections.remove(regID);
if (0 == myIdentifiedConnections.size()) {
myIdentifiedConnections = null;
}
return ret;
}
/** Remove an entry from mySuspendedConnections
*
* @param regID is the VatID to remove.
*/
private VatTPConnection /*nullOK*/ removeSuspended(String regID) {
if (null == mySuspendedConnections) return null;
VatTPConnection ret
= (VatTPConnection)mySuspendedConnections.remove(regID);
if (0 == mySuspendedConnections.size()) {
mySuspendedConnections = null;
}
return ret;
}
/** Remove an entry from mySuspendingConnections
*
* @param regID is the VatID to remove.
*/
private VatTPConnection /*nullOK*/ removeSuspending(String regID) {
if (null == mySuspendingConnections) return null;
VatTPConnection ret
= (VatTPConnection)mySuspendingConnections.remove(regID);
if (0 == mySuspendingConnections.size()) {
mySuspendingConnections = null;
}
return ret;
}
/** Remove an entry from myUnidentifiedConnections
*
* @param regID is the VatID to remove.
*/
private Object /*nullOK*/ removeUnidentified(DataPath path) {
if (null == myUnidentifiedConnections) return null;
Object ret = myUnidentifiedConnections.remove(path);
if (0 == myUnidentifiedConnections.size()) {
myUnidentifiedConnections = null;
}
return ret;
}
/**
* Returns the VatTP configuration parameters as they exist after
* allocating a listen port.
*/
public NetConfig getNetConfig() { return myNetConfig; }
/**
* Return the search path to our own vat.
*/
public ConstList searchPath() {
return myNetConfig.getSearchPath();
}
/**
* Set the object which will respond to vat location queries on our
* listen port.
*
* @param vls The VLS object
*/
public void setVatLocationLookup(VatLocationLookup vls) {
if (myVLS != null) {
Assertion.fail("duplicate setVatLocationLookup");
}
myVLS = vls;
}
/**
* Inform the world that a new conection is ready to pass data. This
* method is called by VatTPConnection after the start up protocol has
* successfully completed.
*
* @param connection The VatTPConnection which is now in RUNNING state.
* @param remoteVatID is the vatID of the remote vat.
* @param isResuming is true if this connection is resuming, false if
* it is starting.
*/
/*package*/ void startupSuccessful(VatTPConnection connection,
String remoteVatID,
boolean isResuming)
{
Object dc = null;
if (isResuming) {
dc = removeSuspended(remoteVatID);
} else {
dc =removeIdentified(remoteVatID);
}
if (null == dc) {
Trace.comm.errorm(connection + " Not registered as pending\n"
+ this);
}
VatTPConnection oldconn
= (VatTPConnection)myRunningDataConnections.put(
remoteVatID, connection);
if (null != oldconn) {
Trace.comm.errorm("Overlaying " + oldconn + " with " + connection);
// try {
oldconn.shutDownConnection(new ConnectionShutDownException(
"Replacing old connection with newly started one"));
// } catch(IOException e) {
// Trace.comm.errorm("Error shutting down connection", e);
// }
}
if (null != myReactor && !isResuming) {
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("call NewConnectionReactor " + myReactor
+ " with " + connection);
}
myReactor.reactToNewConnection(connection);
}
}
/**
* Convert the state of the world to a string and print it.
*
* @return is a String which descirbes this VatTPMgr and the
* state of all the connections it is managing.
*/
public String toString() {
StringBuffer buf = new StringBuffer(500);
buf.append(super.toString()).append("\nlistening at ");
buf.append(myListenThread.listenAddress().toString());
buf.append(" myVatID=").append(myLocalVatID);
if (!myRunner.isCurrentThreadInVat()) {
buf.append("\n***Warning: Caller does not hold vat lock***");
}
if (null != myUnidentifiedConnections) {
buf.append("\n" + myUnidentifiedConnections.size()
+ " Unidentified Connections\n");
listState(myUnidentifiedConnections.keys(), buf);
}
if (null != myIdentifiedConnections) {
buf.append("\n" + myIdentifiedConnections.size()
+ " Identified Starting Connections\n");
listState(myIdentifiedConnections.elements(), buf);
}
if (0 != myRunningDataConnections.size()) {
buf.append("\n" + myRunningDataConnections.size()
+ " Running Connections\n");
listState(myRunningDataConnections.elements(), buf);
}
if (null != mySuspendingConnections) {
buf.append("\n" + mySuspendingConnections.size()
+ " Suspending Connections\n");
listState(mySuspendingConnections.elements(), buf);
}
if (null != mySuspendedConnections) {
buf.append("\n" + mySuspendedConnections.size()
+ " Suspended Connections\n");
listState(mySuspendedConnections.elements(), buf);
}
if (null != myDieingConnections) {
buf.append("\n" + myDieingConnections.size()
+ " Dieing Connections\n");
listState(myDieingConnections.keys(), buf);
}
return buf.toString();
}
/**
* Note that a DataPath for an connection has died before the
* vat at the other end was identified.
*
* @param path is the DataPath object whose connection has died.
*/
/*package*/ void unidentifiedConnectionDied(DataPath path,
String msg,
Throwable problem)
{
String remoteAddr = path.getRemoteAddress();
Object dp = removeUnidentified(path);
if (null == dp) {
Trace.comm.errorm("Dieing DataPath=" + path + " not registered",
new Throwable("locator"));
dp = this; // Something which isn't a resolver
}
if (null != problem) {
if (problem instanceof IOException) {
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm(problem
+ " on unidentified connection to "
+ remoteAddr);
}
} else {
Trace.comm.errorm("Exception on unidentified connection to "
+ remoteAddr + "\n " + problem);
}
} else if (null != msg) {
if (Trace.comm.event && Trace.ON) {
Trace.comm.eventm("Unidentified connection to " + remoteAddr
+ " died, " + msg);
}
problem = new Throwable(msg);
} else {
problem = new Throwable("Undetermined reason");
}
if (dp instanceof Resolver) {
((Resolver)dp).smash(problem);
}
}
}