[INTERNAL][CORE] Adds stream access

This patch introduces new classes and interface. This is necessary to
distinguish between plain stream connections and internal stuff.

In general this patch is a whole rewrite of the DTM.

It now manages two connection pools. One for packet connections and one
for stream connections.

In addition this commits adds Javadoc, TODOs and FIXMEs.

Author: srossbach

core/src/saros/net/IConnectionManager.java

@@ -1,12 +1,26 @@
 package saros.net;
 
 import java.io.IOException;
-import saros.net.internal.IByteStreamConnection;
 import saros.net.stream.StreamMode;
-import saros.net.xmpp.JID;
-
-// TODO Javadoc
 
+/**
+ * The connection manager is responsible for establishing connections to remote addresses.
+ *
+ * <p>It offers support for two connection types.
+ *
+ * <ol>
+ *   <li>Establishing a connection using {@link #connect(String, Object)} to establish a connection
+ *       which is needed in conjunction with the {@link ITransmitter transmitter} and {@link
+ *       IReceiver receiver} in order to send and receive related packages.
+ *   <li>Establishing a connection using {@link #connectStream(String, Object)} to establish a
+ *       {@link IStreamConnection connection} that can be used for custom purposes. In order to get
+ *       notified about such a connection on the remote side you have to install a {@link
+ *       IStreamConnectionListener}.
+ * </ol>
+ *
+ * <p><b>Note</b>: Stream connections must be closed by calling {@link IStreamConnection#close()
+ * close} on the given connection.
+ */
 public interface IConnectionManager {
 
   public static final int IBB_SERVICE = 1;
@@ -21,24 +35,40 @@ public interface IConnectionManager {
    */
   public void setServices(int serviceMask);
 
-  /** @deprecated */
-  @Deprecated
-  public IByteStreamConnection connect(JID peer) throws IOException;
+  public void addStreamConnectionListener(final IStreamConnectionListener listener);
 
-  public IByteStreamConnection connect(String connectionID, JID peer) throws IOException;
+  public void removeStreamConnectionListener(final IStreamConnectionListener listener);
 
   /**
-   * @deprecated Disconnects {@link IByteStreamConnection} with the specified peer
-   * @param peer {@link JID} of the peer to disconnect the {@link IByteStreamConnection}
+   * Connects to the given address using the given stream ID.
+   *
+   * @param id the ID of the stream
+   * @param address the remote address to connect to
+   * @return the stream connection
+   * @throws IOException if an I/O error occurs or such a stream already exists
    */
-  @Deprecated
-  public boolean closeConnection(JID peer);
+  public IStreamConnection connectStream(String id, Object address) throws IOException;
 
-  public boolean closeConnection(String connectionIdentifier, JID peer);
+  /**
+   * Connects to the given address using the given ID.
+   *
+   * @param id
+   * @param address
+   * @throws IOException if an I/O error occurs
+   */
+  public void connect(String id, Object address) throws IOException;
 
-  /** @deprecated */
-  @Deprecated
-  public StreamMode getTransferMode(JID jid);
+  /**
+   * Closes the given connection.
+   *
+   * @param id the ID of the connection
+   * @param address the remote address
+   * @return <code>true</code> if the connection was closed, <code>false</code> if no such
+   *     connection exists
+   */
+  // TODO rename to close
+  public boolean closeConnection(String id, Object address);
 
-  public StreamMode getTransferMode(String connectionID, JID jid);
+  // TODO RENAME
+  public StreamMode getTransferMode(String connectionID, Object address);
 }

core/src/saros/net/IStreamConnection.java

@@ -0,0 +1,44 @@
+package saros.net;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import saros.net.internal.IConnection;
+
+/**
+ * A stream connection consists of an input and out stream. It up to the caller to gracefully
+ * shutdown the connections.
+ */
+public interface IStreamConnection extends IConnection {
+
+  /**
+   * Gets the input stream to receive data.
+   *
+   * @return the input stream to receive data
+   * @throws IOException if an I/O error occurs
+   */
+  public InputStream getInputStream() throws IOException;
+
+  /**
+   * Gets the output stream to send data.
+   *
+   * @return the output stream to send data
+   * @throws IOException if an I/O error occurs
+   */
+  public OutputStream getOutputStream() throws IOException;
+
+  /**
+   * Returns the read timeout in milliseconds.
+   *
+   * @return the read timeout
+   * @throws IOException if an I/O error occurs
+   */
+  public int getReadTimeout() throws IOException;
+
+  /**
+   * Sets the read timeout in milliseconds.
+   *
+   * @throws IOException if an I/O error occurs
+   */
+  public void setReadTimeout(int timeout) throws IOException;
+}

core/src/saros/net/IStreamConnectionListener.java

@@ -0,0 +1,23 @@
+package saros.net;
+
+/**
+ * Listener interface for accepting remote stream connections.
+ *
+ * <p>The listener must either accept or reject the connection. If there are multiple listeners
+ * installed via {@link IConnectionManager#addStreamConnectionListener(IStreamConnectionListener)}
+ * the connection will only be rejected if all listeners reject.
+ *
+ * <p>Furthermore if a listener accepts the request every outstanding listener will <b>not</b> get
+ * notified about this connection establishment.
+ */
+public interface IStreamConnectionListener {
+
+  /**
+   * Gets called when a connection was established.
+   *
+   * @param connection the established connection
+   * @return <code>true</code> to accept the connection or <code>false</code> to reject the
+   *     connection.
+   */
+  public boolean connectionEstablished(IStreamConnection connection);
+}

core/src/saros/net/ITransmitter.java

@@ -1,37 +1,35 @@
-/*
- * DPP - Serious Distributed Pair Programming (c) Freie Universität Berlin -
- * Fachbereich Mathematik und Informatik - 2006 (c) Riad Djemili - 2006
- *
- * This program is free software; you can redistribute it and/or modify it under
- * the terms of the GNU General Public License as published by the Free Software
- * Foundation; either version 1, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful, but WITHOUT
- * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
- * FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
- * details.
- *
- * You should have received a copy of the GNU General Public License along with
- * this program; if not, write to the Free Software Foundation, Inc., 675 Mass
- * Ave, Cambridge, MA 02139, USA.
- */
-
 package saros.net;
 
 import java.io.IOException;
 import org.jivesoftware.smack.packet.Packet;
 import org.jivesoftware.smack.packet.PacketExtension;
-import saros.annotations.Component;
 import saros.net.xmpp.JID;
 
 /**
- * A humble interface that is responsible for network functionality. The idea behind this interface
- * is to only encapsulates the least possible amount of functionality - the one that can't be easily
- * tested.
+ * Interface for sending packets to remote addresses. In general this interface offers two
+ * possibilities for sending packets.
  *
- * @author rdjemili
+ * <ol>
+ *   <li>Either sending packets using the default network environment by calling {@link
+ *       #sendPacket(Packet)} or {@link #sendPacketExtension(JID, PacketExtension)}.
+ *   <li>Using {@link #send(String, JID, PacketExtension)} using a specific connection that must
+ *       first be established by calling {@link IConnectionManager#connect(String, Object)}.
+ * </ol>
+ *
+ * The second option should always be used as default option when sending packets frequently and
+ * over a longer time span to an already known address.
+ *
+ * <p><b>Implementation notes</b>: Implementation should consider to support connection ID's through
+ * the {@link IConnectionManager}. If this is not possible the implementation <b>must</b> ensure
+ * that connection lost is properly detected, i.e sending packets to a server which may route the
+ * packets at a later time without getting an acknowledgement if the packet has been received is a
+ * <b>violation</b> of the contract.
+ *
+ * @see IConnectionManager
+ */
+/*
+ * TODO ensure we use IQ packets so the server must return an error. Afterwards we can change the contract of this interface.
  */
-@Component(module = "net")
 public interface ITransmitter {
 
   /**
@@ -63,13 +61,14 @@ public interface ITransmitter {
   public void send(JID recipient, PacketExtension extension) throws IOException;
 
   /**
-   * Sends the given {@link PacketExtension} to the given {@link JID} using a direct stream
-   * connection. The connection must be already established to the recipient with the given id.
+   * Sends the given {@link PacketExtension} to the given {@link JID} using the given connection ID.
+   * A connection with the given connection id must already been established.
    *
-   * @param connectionID the id of the connection
+   * @param connectionID the ID of the connection
    * @param recipient the recipient of the extension
    * @param extension the extension to send
    * @throws IOException if an I/O error occurs
+   * @see IConnectionManager#connect(String, Object)
    */
   public void send(String connectionID, JID recipient, PacketExtension extension)
       throws IOException;

core/src/saros/net/internal/BinaryChannelConnection.java

@@ -30,7 +30,7 @@
  * @author coezbek
  * @author srossbach
  */
-public class BinaryChannelConnection implements IByteStreamConnection {
+public class BinaryChannelConnection implements IPacketConnection {
 
   private static final Logger LOG = Logger.getLogger(BinaryChannelConnection.class);
 
@@ -49,21 +49,23 @@ private static class Opcode {
   /** Max size of data chunks */
   private static final int CHUNKSIZE = 32 * 1024 - 1;
 
-  private IByteStreamConnectionListener listener;
   private ReceiverThread receiveThread;
 
   private final JID remoteAddress;
   private final JID localAddress;
 
   private final String connectionID;
 
+  private final IConnectionClosedCallback callback;
+
   private IDPool idPool = new IDPool();
 
   private boolean connected;
   private boolean initialized;
 
   private Map<Integer, ByteArrayOutputStream> pendingFragmentedPackets =
       new HashMap<Integer, ByteArrayOutputStream>();
+
   private Map<Integer, BinaryXMPPExtension> pendingXMPPExtensions =
       new HashMap<Integer, BinaryXMPPExtension>();
 
@@ -111,24 +113,16 @@ public void run() {
 
   private IBinaryXMPPExtensionReceiver receiver;
 
-  public BinaryChannelConnection(
-      JID localAddress,
-      JID remoteAddress,
-      String connectionID,
-      ByteStream stream,
-      StreamMode mode,
-      IByteStreamConnectionListener listener)
-      throws IOException {
-    this.listener = listener;
-    this.localAddress = localAddress;
-    this.remoteAddress = remoteAddress;
-    this.connectionID = connectionID;
-    this.stream = stream;
-    this.stream.setReadTimeout(0); // keep connection alive
-    this.mode = mode;
+  public BinaryChannelConnection(ByteStream stream, IConnectionClosedCallback callback) {
+    this.callback = callback;
+    // FIXME
+    this.localAddress = (JID) stream.getLocalAddress();
+    // FIXME
+    this.remoteAddress = (JID) stream.getRemoteAddress();
 
-    outputStream = new DataOutputStream(new BufferedOutputStream(stream.getOutputStream()));
-    inputStream = new DataInputStream(new BufferedInputStream(stream.getInputStream()));
+    this.connectionID = stream.getId();
+    this.mode = stream.getMode();
+    this.stream = stream;
   }
 
   @Override
@@ -138,14 +132,17 @@ public void setBinaryXMPPExtensionReceiver(IBinaryXMPPExtensionReceiver receiver
     this.receiver = receiver;
   }
 
-  @Override
-  public synchronized void initialize() {
+  public synchronized void initialize() throws IOException {
     if (initialized) return;
 
     /*
      * it is ok to start the receiver a bit later because the data will be
      * already buffered by SMACK or the OS
      */
+    stream.setReadTimeout(0); // keep connection alive
+    outputStream = new DataOutputStream(new BufferedOutputStream(stream.getOutputStream()));
+    inputStream = new DataInputStream(new BufferedInputStream(stream.getInputStream()));
+
     receiveThread = new ReceiverThread();
     receiveThread.setName("BinaryChannel-" + remoteAddress.getName());
     receiveThread.start();
@@ -154,13 +151,23 @@ public synchronized void initialize() {
   }
 
   @Override
-  public String getConnectionID() {
-    return connectionID;
+  public Object getLocalAddress() {
+    return localAddress;
   }
 
   @Override
-  public synchronized boolean isConnected() {
-    return connected;
+  public JID getRemoteAddress() {
+    return remoteAddress;
+  }
+
+  @Override
+  public StreamMode getMode() {
+    return mode;
+  }
+
+  @Override
+  public String getId() {
+    return connectionID;
   }
 
   @Override
@@ -192,17 +199,7 @@ public void close() {
       }
     }
 
-    listener.connectionClosed(connectionID, this);
-  }
-
-  @Override
-  public StreamMode getMode() {
-    return mode;
-  }
-
-  @Override
-  public JID getRemoteAddress() {
-    return remoteAddress;
+    if (callback != null) callback.connectionClosed(this);
   }
 
   @Override
@@ -436,6 +433,10 @@ private BinaryXMPPExtension readNextXMPPExtension() throws IOException {
     throw new InterruptedIOException("interrupted while reading stream data");
   }
 
+  private synchronized boolean isConnected() {
+    return connected;
+  }
+
   private synchronized void sendData(int fragmentId, byte[] data, int offset, int length)
       throws IOException {
 
@@ -485,7 +486,15 @@ private void splitAndSend(byte[] data, int chunks, int fragmentId) throws IOExce
 
   @Override
   public String toString() {
-    return "[mode=" + getMode() + ", id=" + connectionID + "]" + " " + remoteAddress;
+    return "PacketConnection [id="
+        + getId()
+        + ", mode="
+        + getMode()
+        + ", localAddress="
+        + getLocalAddress()
+        + ", remoteAddress="
+        + getRemoteAddress()
+        + "]";
   }
 
   static class IDPool {