[INTERNAL][CORE] Prepares Stream access

This patch introduces an new class / interface which is just used for
the ITransmitter and IReceiver. This is necessary to distinguish between
plan stream connections and internal stuff.

Author: srossbach

core/src/saros/net/IConnectionManager.java

@@ -1,7 +1,7 @@
 package saros.net;
 
 import java.io.IOException;
-import saros.net.internal.IByteStreamConnection;
+import saros.net.internal.IConnection;
 import saros.net.stream.StreamMode;
 import saros.net.xmpp.JID;
 
@@ -21,24 +21,29 @@ public interface IConnectionManager {
    */
   public void setServices(int serviceMask);
 
+  public void addStreamConnectionListener(final IStreamConnectionListener listener);
+
+  public void removeStreamConnectionListener(final IStreamConnectionListener listener);
+
+  public IStreamConnection connectStream(String id, Object address) throws IOException;
   /** @deprecated */
   @Deprecated
-  public IByteStreamConnection connect(JID peer) throws IOException;
+  public IConnection connect(Object address) throws IOException;
 
-  public IByteStreamConnection connect(String connectionID, JID peer) throws IOException;
+  public IConnection connect(String connectionID, Object address) throws IOException;
 
   /**
-   * @deprecated Disconnects {@link IByteStreamConnection} with the specified peer
-   * @param peer {@link JID} of the peer to disconnect the {@link IByteStreamConnection}
+   * @deprecated Disconnects with the specified address
+   * @param address
    */
   @Deprecated
-  public boolean closeConnection(JID peer);
+  public boolean closeConnection(Object address);
 
-  public boolean closeConnection(String connectionIdentifier, JID peer);
+  public boolean closeConnection(String connectionIdentifier, Object address);
 
   /** @deprecated */
   @Deprecated
   public StreamMode getTransferMode(JID jid);
 
-  public StreamMode getTransferMode(String connectionID, JID jid);
+  public StreamMode getTransferMode(String connectionID, Object address);
 }

core/src/saros/net/IStreamConnection.java

@@ -0,0 +1,17 @@
+package saros.net;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import saros.net.internal.IConnection;
+
+public interface IStreamConnection extends IConnection {
+
+  public InputStream getInputStream() throws IOException;
+
+  public OutputStream getOutputStream() throws IOException;
+
+  public int getReadTimeout() throws IOException;
+
+  public void setReadTimeout(int timeout) throws IOException;
+}

core/src/saros/net/IStreamConnectionListener.java

@@ -0,0 +1,6 @@
+package saros.net;
+
+public interface IStreamConnectionListener {
+
+  public boolean streamConnectionEstablished(String id, IStreamConnection connection);
+}

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,13 @@ public synchronized void initialize() {
   }
 
   @Override
-  public String getConnectionID() {
-    return connectionID;
+  public Object getLocalAddress() {
+    return stream.getRemoteAddress();
   }
 
   @Override
-  public synchronized boolean isConnected() {
-    return connected;
+  public String getId() {
+    return connectionID;
   }
 
   @Override
@@ -192,10 +189,9 @@ public void close() {
       }
     }
 
-    listener.connectionClosed(connectionID, this);
+    if (callback != null) callback.connectionClosed(this);
   }
 
-  @Override
   public StreamMode getMode() {
     return mode;
   }
@@ -436,6 +432,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 {
 

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

@@ -15,8 +15,7 @@ final class ConnectionPool {
 
   private boolean isOpen;
 
-  private final Map<String, IByteStreamConnection> pool =
-      new HashMap<String, IByteStreamConnection>();
+  private final Map<String, IConnection> pool = new HashMap<>();
 
   /**
    * Opens the connection pool. After the connection pool is opened connections can be added,
@@ -31,19 +30,19 @@ public synchronized void open() {
    */
   public void close() {
 
-    final Map<String, IByteStreamConnection> currentPoolCopy;
+    final Map<String, IConnection> currentPoolCopy;
 
     synchronized (this) {
       if (!isOpen) return;
 
       isOpen = false;
-      currentPoolCopy = new HashMap<String, IByteStreamConnection>(pool);
+      currentPoolCopy = new HashMap<>(pool);
       pool.clear();
     }
 
-    for (Entry<String, IByteStreamConnection> entry : currentPoolCopy.entrySet()) {
+    for (Entry<String, IConnection> entry : currentPoolCopy.entrySet()) {
       final String id = entry.getKey();
-      final IByteStreamConnection connection = entry.getValue();
+      final IConnection connection = entry.getValue();
       connection.close();
 
       LOG.debug("closed connection [id=" + id + "]: " + connection);
@@ -57,7 +56,7 @@ public void close() {
    * @return the connection associated with the id or <code>null</code> if no such connection exists
    *     or the pool is closed
    */
-  public synchronized IByteStreamConnection get(final String id) {
+  public synchronized IConnection get(final String id) {
     return pool.get(id);
   }
 
@@ -70,8 +69,7 @@ public synchronized IByteStreamConnection get(final String id) {
    *     that was already added with the given id or <code>null</code> if no connection was added
    *     with the given id
    */
-  public synchronized IByteStreamConnection add(
-      final String id, final IByteStreamConnection connection) {
+  public synchronized IConnection add(final String id, final IConnection connection) {
 
     if (!isOpen) return connection;
 
@@ -85,7 +83,7 @@ public synchronized IByteStreamConnection add(
    * @return the connection associated with the id or <code>null</code> if no such connection exists
    *     or the pool is closed
    */
-  public synchronized IByteStreamConnection remove(final String id) {
+  public synchronized IConnection remove(final String id) {
     if (!isOpen) return null;
 
     return pool.remove(id);