[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/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);
 
@@ -111,24 +111,16 @@ public void run() {
 
   private IBinaryXMPPExtensionReceiver receiver;
 
-  public BinaryChannelConnection(
-      JID localAddress,
-      JID remoteAddress,
-      String connectionID,
-      ByteStream stream,
-      StreamMode mode,
-      IByteStreamConnectionListener listener)
-      throws IOException {
+  public BinaryChannelConnection(ByteStream stream, IByteStreamConnectionListener listener) {
     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;
+    // 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
@@ -139,13 +131,17 @@ public void setBinaryXMPPExtensionReceiver(IBinaryXMPPExtensionReceiver 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();

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

@@ -18,7 +18,9 @@
 import saros.context.IContextKeyBindings.Socks5StreamService;
 import saros.net.ConnectionState;
 import saros.net.IConnectionManager;
+import saros.net.stream.ByteStream;
 import saros.net.stream.IStreamService;
+import saros.net.stream.IStreamServiceListener;
 import saros.net.stream.StreamMode;
 import saros.net.xmpp.IConnectionListener;
 import saros.net.xmpp.JID;
@@ -28,17 +30,14 @@
 /**
  * This class is responsible for handling all transfers of binary data. It maintains a map of
  * established connections and tries to reuse them.
- *
- * @author srossbach
- * @author coezbek
- * @author jurke
  */
 @Component(module = "net")
 public class DataTransferManager implements IConnectionListener, IConnectionManager {
 
   private static final Logger LOG = Logger.getLogger(DataTransferManager.class);
 
-  private static final String DEFAULT_CONNECTION_ID = "default";
+  // package private for testing purposes
+  static final String DEFAULT_CONNECTION_ID = "default";
 
   private static final String IN = "in";
 
@@ -65,6 +64,12 @@ public class DataTransferManager implements IConnectionListener, IConnectionMana
   private final CopyOnWriteArrayList<IByteStreamConnectionListener> connectionListeners =
       new CopyOnWriteArrayList<>();
 
+  private final CopyOnWriteArrayList<IPacketConnectionListener> packetConnectionListeners =
+      new CopyOnWriteArrayList<>();
+
+  private final IStreamServiceListener streamServiceListener =
+      (stream) -> createAndAnnouncePacketConnection(stream, true);
+
   private final IByteStreamConnectionListener byteStreamConnectionListener =
       new IByteStreamConnectionListener() {
 
@@ -81,11 +86,6 @@ public void connectionChanged(
               toConnectionIDToken(
                   connectionId, incomingRequest ? IN : OUT, connection.getRemoteAddress());
 
-          /// TODO we currently have to announce not initialized connections otherwise the IReceiver
-          // will miss updates
-
-          notfiyconnectionChanged(id, connection, incomingRequest);
-
           LOG.debug(
               "bytestream connection changed "
                   + connection
@@ -123,7 +123,16 @@ public void connectionChanged(
             }
           }
 
-          connection.initialize();
+          try {
+            connection.initialize();
+          } catch (IOException e) {
+            LOG.error("failed to initialize connection: " + connection);
+            connection.close();
+            connectionPool.remove(id);
+            return;
+          }
+
+          notfiyconnectionChanged(id, connection, incomingRequest);
         }
 
         @Override
@@ -234,6 +243,14 @@ public void removeConnectionListener(final IByteStreamConnectionListener listene
     connectionListeners.remove(listener);
   }
 
+  public void addPacketConnectionListener(final IPacketConnectionListener listener) {
+    packetConnectionListeners.addIfAbsent(listener);
+  }
+
+  public void removePacketConnectionListener(final IPacketConnectionListener listener) {
+    packetConnectionListeners.remove(listener);
+  }
+
   private IByteStreamConnection connectInternal(String connectionID, JID peer) throws IOException {
 
     IByteStreamConnection connection = null;
@@ -265,6 +282,8 @@ private IByteStreamConnection connectInternal(String connectionID, JID peer) thr
       final ArrayList<IStreamService> currentStreamServices =
           new ArrayList<IStreamService>(streamServices);
 
+      ByteStream byteStream = null;
+
       for (IStreamService streamService : currentStreamServices) {
         LOG.info(
             "establishing connection to "
@@ -274,7 +293,7 @@ private IByteStreamConnection connectInternal(String connectionID, JID peer) thr
                 + " using stream service "
                 + streamService);
         try {
-          connection = streamService.connect(connectionID, peer);
+          byteStream = streamService.connect(connectionID, peer);
           break;
         } catch (IOException e) {
           LOG.warn("failed to connect to " + peer + " using stream service: " + streamService, e);
@@ -298,9 +317,8 @@ private IByteStreamConnection connectInternal(String connectionID, JID peer) thr
         }
       }
 
-      if (connection != null) {
-        byteStreamConnectionListener.connectionChanged(connectionID, connection, false);
-
+      if (byteStream != null) {
+        connection = createAndAnnouncePacketConnection(byteStream, false);
         return connection;
       }
 
@@ -347,7 +365,7 @@ private void prepareConnection(final Connection connection) {
     connectionPool.open();
 
     for (IStreamService streamService : streamServices)
-      streamService.initialize(xmppConnection, byteStreamConnectionListener);
+      streamService.initialize(xmppConnection, streamServiceListener);
   }
 
   private void disposeConnection() {
@@ -431,4 +449,22 @@ private void notfiyconnectionChanged(
       }
     }
   }
+
+  private IByteStreamConnection createAndAnnouncePacketConnection(
+      final ByteStream byteStream, final boolean isIncoming) {
+    final IPacketConnection connection =
+        new BinaryChannelConnection(byteStream, byteStreamConnectionListener);
+
+    for (final IPacketConnectionListener listener : packetConnectionListeners) {
+      try {
+        listener.connectionEstablished(connection);
+      } catch (RuntimeException e) {
+        LOG.error("invoking connectionEstablished() on listener: " + listener + " failed", e);
+      }
+    }
+
+    byteStreamConnectionListener.connectionChanged(byteStream.getId(), connection, isIncoming);
+
+    return connection;
+  }
 }

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

@@ -17,17 +17,7 @@ public interface IByteStreamConnection {
    * Initializes the byte stream connection. After the initialization is performed the byte stream
    * connection must be able to send and receive data.
    */
-  public void initialize();
-
-  /**
-   * If this call returns the data has been send successfully, otherwise an IOException is thrown
-   * with the reason why the transfer failed.
-   *
-   * @param data The data to be sent.
-   * @throws IOException if the send failed
-   * @blocking Send the given data as a blocking operation.
-   */
-  public void send(TransferDescription data, byte[] content) throws IOException;
+  public void initialize() throws IOException;
 
   /**
    * Returns the connection id of this connection.
@@ -37,6 +27,4 @@ public interface IByteStreamConnection {
   public String getConnectionID();
 
   public StreamMode getMode();
-
-  public void setBinaryXMPPExtensionReceiver(IBinaryXMPPExtensionReceiver receiver);
 }

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

@@ -0,0 +1,22 @@
+package saros.net.internal;
+
+import java.io.IOException;
+import saros.net.IReceiver;
+import saros.net.ITransmitter;
+
+/**
+ * A packet connection is internally used as an abstraction layer to coordinate the transmission of
+ * and receiving of packets between the {@link IReceiver} and {@link ITransmitter}.
+ */
+public interface IPacketConnection extends IByteStreamConnection {
+
+  /** Sends the given data along with the given description. */
+  public void send(TransferDescription description, byte[] data) throws IOException;
+
+  /**
+   * Sets the receiver for incoming {@link BinaryXMPPExtension packet extensions}.
+   *
+   * @param receiver
+   */
+  public void setBinaryXMPPExtensionReceiver(IBinaryXMPPExtensionReceiver receiver);
+}

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

@@ -0,0 +1,13 @@
+package saros.net.internal;
+
+/** Listener interface for {@link IPacketConnection packet connections}. */
+public interface IPacketConnectionListener {
+
+  /**
+   * Gets called when a new packet connection was established. The connection is <b>not</b>
+   * initialized at this point.
+   *
+   * @param connection
+   */
+  public void connectionEstablished(IPacketConnection connection);
+}

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

@@ -89,13 +89,10 @@ public XMPPReceiver(
     this.parser = new MXParser();
 
     connectionService.addListener(connectionListener);
-    dataTransferManager.addConnectionListener(
-        new IByteStreamConnectionListener() {
+    dataTransferManager.addPacketConnectionListener(
+        new IPacketConnectionListener() {
           @Override
-          public void connectionChanged(
-              final String connectionId,
-              final IByteStreamConnection connection,
-              final boolean incomingRequest) {
+          public void connectionEstablished(final IPacketConnection connection) {
             connection.setBinaryXMPPExtensionReceiver(XMPPReceiver.this);
           }
         });

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

@@ -91,6 +91,10 @@ public void send(String connectionID, JID recipient, PacketExtension extension)
 
     if (connection == null) connection = dataManager.connect(recipient);
 
+    if (!(connection instanceof IPacketConnection))
+      throw new IOException(
+          "connection id '" + connectionID + "' references an unmanaged stream connection");
+
     /*
      * The TransferDescription can be created out of the session, the name
      * and namespace of the packet extension and standard values and thus
@@ -109,7 +113,7 @@ public void send(String connectionID, JID recipient, PacketExtension extension)
       transferDescription.setCompressContent(true);
     }
 
-    sendPacketExtension(connection, transferDescription, data);
+    sendPacketExtension((IPacketConnection) connection, transferDescription, data);
   }
 
   @Override
@@ -189,7 +193,7 @@ private synchronized boolean isConnectionInvalid() {
   }
 
   private void sendPacketExtension(
-      final IByteStreamConnection connection, final TransferDescription description, byte[] payload)
+      final IPacketConnection connection, final TransferDescription description, byte[] payload)
       throws IOException {
 
     boolean sendPacket = true;

core/src/saros/net/stream/ByteStream.java

@@ -16,4 +16,12 @@ public interface ByteStream {
   public int getReadTimeout() throws IOException;
 
   public void setReadTimeout(int timeout) throws IOException;
+
+  public Object getLocalAddress();
+
+  public Object getRemoteAddress();
+
+  public StreamMode getMode();
+
+  public String getId();
 }