[INTERNAL][CORE] Moves Receiving logic to IReceiver

This patch moves the receiving logic to the logic where it should belong
to.

Author: srossbach

core/src/saros/net/IConnectionManager.java

@@ -21,10 +21,6 @@ public interface IConnectionManager {
    */
   public void setServices(int serviceMask);
 
-  public void addTransferListener(ITransferListener listener);
-
-  public void removeTransferListener(ITransferListener listener);
-
   /** @deprecated */
   @Deprecated
   public IByteStreamConnection connect(JID peer) throws IOException;

core/src/saros/net/IReceiver.java

@@ -3,7 +3,6 @@
 import org.jivesoftware.smack.PacketListener;
 import org.jivesoftware.smack.filter.PacketFilter;
 import org.jivesoftware.smack.packet.Packet;
-import saros.net.internal.BinaryXMPPExtension;
 
 public interface IReceiver {
 
@@ -52,9 +51,6 @@ public interface IReceiver {
    */
   public PacketCollector createCollector(PacketFilter filter);
 
-  /** FOR INTERNAL USE */
-  public void processBinaryXMPPExtension(BinaryXMPPExtension extension);
-
   public default void addTransferListener(ITransferListener listener) {
     // NOP
   }

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

@@ -90,7 +90,10 @@ public void run() {
 
       LOG.debug(connection + " ReceiverThread started.");
       try {
-        while (!isInterrupted()) listener.receive(readNextXMPPExtension());
+        while (!isInterrupted()) {
+          final BinaryXMPPExtension extension = readNextXMPPExtension();
+          if (receiver != null) receiver.receive(extension);
+        }
 
       } catch (SocketException e) {
         LOG.debug(connection + " connection closed locally: " + e.getMessage());
@@ -106,6 +109,8 @@ public void run() {
     }
   }
 
+  private IBinaryXMPPExtensionReceiver receiver;
+
   public BinaryChannelConnection(
       JID localAddress,
       JID remoteAddress,
@@ -126,6 +131,13 @@ public BinaryChannelConnection(
     inputStream = new DataInputStream(new BufferedInputStream(stream.getInputStream()));
   }
 
+  @Override
+  public void setBinaryXMPPExtensionReceiver(IBinaryXMPPExtensionReceiver receiver) {
+    if (this.receiver != null || receiver == null) return;
+
+    this.receiver = receiver;
+  }
+
   @Override
   public synchronized void initialize() {
     if (initialized) return;

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

@@ -1,6 +1,5 @@
 package saros.net.internal;
 
-import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InterruptedIOException;
 import java.util.ArrayList;
@@ -12,18 +11,13 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
-import java.util.zip.DataFormatException;
-import java.util.zip.Inflater;
 import org.apache.log4j.Logger;
 import org.jivesoftware.smack.Connection;
 import saros.annotations.Component;
 import saros.context.IContextKeyBindings.IBBStreamService;
 import saros.context.IContextKeyBindings.Socks5StreamService;
 import saros.net.ConnectionState;
 import saros.net.IConnectionManager;
-import saros.net.IPacketInterceptor;
-import saros.net.IReceiver;
-import saros.net.ITransferListener;
 import saros.net.stream.IStreamService;
 import saros.net.stream.StreamMode;
 import saros.net.xmpp.IConnectionListener;
@@ -44,28 +38,18 @@ public class DataTransferManager implements IConnectionListener, IConnectionMana
 
   private static final Logger LOG = Logger.getLogger(DataTransferManager.class);
 
-  private static final int CHUNKSIZE = 16 * 1024;
-
   private static final String DEFAULT_CONNECTION_ID = "default";
 
   private static final String IN = "in";
 
   private static final String OUT = "out";
 
-  private final CopyOnWriteArrayList<IPacketInterceptor> packetInterceptors =
-      new CopyOnWriteArrayList<IPacketInterceptor>();
-
-  private final List<ITransferListener> transferListeners =
-      new CopyOnWriteArrayList<ITransferListener>();
-
   private volatile JID currentLocalJID;
 
   private Connection xmppConnection;
 
   private int serviceMask = -1;
 
-  private final IReceiver receiver;
-
   private final IStreamService mainService;
 
   private final IStreamService fallbackService;
@@ -78,57 +62,15 @@ public class DataTransferManager implements IConnectionListener, IConnectionMana
 
   private final List<IStreamService> streamServices = new CopyOnWriteArrayList<IStreamService>();
 
+  private final CopyOnWriteArrayList<IByteStreamConnectionListener> connectionListeners =
+      new CopyOnWriteArrayList<>();
+
   private final IByteStreamConnectionListener byteStreamConnectionListener =
       new IByteStreamConnectionListener() {
 
-        @Override
-        public void receive(final BinaryXMPPExtension extension) {
-
-          boolean dispatchPacket = true;
-
-          for (IPacketInterceptor packetInterceptor : packetInterceptors)
-            dispatchPacket &= packetInterceptor.receivedPacket(extension);
-
-          if (!dispatchPacket) return;
-
-          if (LOG.isTraceEnabled())
-            LOG.trace(
-                "received binary XMPP extension: "
-                    + extension.getTransferDescription()
-                    + ", size: "
-                    + extension.getCompressedSize()
-                    + ", RX time: "
-                    + extension.getTransferDuration()
-                    + " ms ["
-                    + extension.getTransferMode()
-                    + "]");
-
-          if (extension.getTransferDescription().compressContent()) {
-            byte[] payload = extension.getPayload();
-            long compressedPayloadLength = payload.length;
-
-            try {
-              payload = inflate(payload);
-            } catch (IOException e) {
-              LOG.error("could not decompress extension payload", e);
-              return;
-            }
-
-            extension.setPayload(compressedPayloadLength, payload);
-          }
-
-          notifyDataReceived(
-              extension.getTransferMode(),
-              extension.getCompressedSize(),
-              extension.getUncompressedSize(),
-              extension.getTransferDuration());
-
-          receiver.processBinaryXMPPExtension(extension);
-        }
-
         @Override
         public void connectionChanged(
-            final String connectionID,
+            final String connectionId,
             final IByteStreamConnection connection,
             final boolean incomingRequest) {
 
@@ -137,7 +79,12 @@ public void connectionChanged(
 
           final String id =
               toConnectionIDToken(
-                  connectionID, incomingRequest ? IN : OUT, connection.getRemoteAddress());
+                  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 "
@@ -180,35 +127,25 @@ public void connectionChanged(
         }
 
         @Override
-        public void connectionClosed(String connectionID, IByteStreamConnection connection) {
-          closeConnection(connectionID, connection.getRemoteAddress());
+        public void connectionClosed(
+            final String connectionId, final IByteStreamConnection connection) {
+          closeConnection(connectionId, connection.getRemoteAddress());
+          notfiyConnectionClosed(connectionId, connection);
         }
       };
 
   public DataTransferManager(
       XMPPConnectionService connectionService,
-      IReceiver receiver,
       @Nullable @Socks5StreamService IStreamService mainService,
       @Nullable @IBBStreamService IStreamService fallbackService) {
 
-    this.receiver = receiver;
     this.fallbackService = fallbackService;
     this.mainService = mainService;
     this.setStreamServices();
 
     connectionService.addListener(this);
   }
 
-  @Override
-  public void addTransferListener(ITransferListener listener) {
-    transferListeners.add(listener);
-  }
-
-  @Override
-  public void removeTransferListener(ITransferListener listener) {
-    transferListeners.remove(listener);
-  }
-
   /** @deprecated */
   @Override
   @Deprecated
@@ -289,6 +226,14 @@ public StreamMode getTransferMode(String connectionID, JID jid) {
     return connection == null ? StreamMode.NONE : connection.getMode();
   }
 
+  public void addConnectionListener(final IByteStreamConnectionListener listener) {
+    connectionListeners.addIfAbsent(listener);
+  }
+
+  public void removeConnectionListener(final IByteStreamConnectionListener listener) {
+    connectionListeners.remove(listener);
+  }
+
   private IByteStreamConnection connectInternal(String connectionID, JID peer) throws IOException {
 
     IByteStreamConnection connection = null;
@@ -433,27 +378,6 @@ public void connectionStateChanged(Connection connection, ConnectionState newSta
     else if (this.xmppConnection != null) disposeConnection();
   }
 
-  // TODO: move to ITransmitter
-  public void addPacketInterceptor(IPacketInterceptor interceptor) {
-    packetInterceptors.addIfAbsent(interceptor);
-  }
-
-  // TODO: move to IReceiver
-  public void removePacketInterceptor(IPacketInterceptor interceptor) {
-    packetInterceptors.remove(interceptor);
-  }
-
-  /**
-   * Left over and <b>MUST</b> only used by the STF
-   *
-   * @param extension
-   * @deprecated
-   */
-  @Deprecated
-  public void addIncomingTransferObject(BinaryXMPPExtension extension) {
-    byteStreamConnectionListener.receive(extension);
-  }
-
   /**
    * Returns the current connection for the remote side. If the local side is connected to the
    * remote side as well as the remote side is connected to the local side the local to remote
@@ -482,39 +406,29 @@ private static String toConnectionIDToken(String connectionIdentifier, String mo
     return connectionIdentifier + ":" + mode + ":" + jid.toString();
   }
 
-  private void notifyDataReceived(
-      final StreamMode mode,
-      final long sizeCompressed,
-      final long sizeUncompressed,
-      final long duration) {
+  private void notfiyConnectionClosed(
+      final String connectionId, final IByteStreamConnection connection) {
 
-    for (final ITransferListener listener : transferListeners) {
+    for (final IByteStreamConnectionListener listener : connectionListeners) {
       try {
-        listener.received(mode, sizeCompressed, sizeUncompressed, duration);
+        listener.connectionClosed(connectionId, connection);
       } catch (RuntimeException e) {
-        LOG.error("invoking received() on listener: " + listener + " failed", e);
+        LOG.error("invoking connectionClosed() on listener: " + listener + " failed", e);
       }
     }
   }
 
-  private static byte[] inflate(byte[] input) throws IOException {
-
-    ByteArrayOutputStream bos;
-    Inflater decompressor = new Inflater();
+  private void notfiyconnectionChanged(
+      final String connectionId,
+      final IByteStreamConnection connection,
+      final boolean incomingRequest) {
 
-    decompressor.setInput(input, 0, input.length);
-    bos = new ByteArrayOutputStream(input.length);
-
-    byte[] buf = new byte[CHUNKSIZE];
-
-    try {
-      while (!decompressor.finished()) {
-        int count = decompressor.inflate(buf);
-        bos.write(buf, 0, count);
+    for (final IByteStreamConnectionListener listener : connectionListeners) {
+      try {
+        listener.connectionChanged(connectionId, connection, incomingRequest);
+      } catch (RuntimeException e) {
+        LOG.error("invoking connectionChanged() on listener: " + listener + " failed", e);
       }
-      return bos.toByteArray();
-    } catch (DataFormatException e) {
-      throw new IOException("failed to inflate data", e);
     }
   }
 }

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

@@ -0,0 +1,11 @@
+package saros.net.internal;
+
+public interface IBinaryXMPPExtensionReceiver {
+
+  /**
+   * Gets called when a {@linkplain BinaryXMPPExtension} was received.
+   *
+   * @param extension
+   */
+  public void receive(final BinaryXMPPExtension extension);
+}

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

@@ -37,4 +37,6 @@ public interface IByteStreamConnection {
   public String getConnectionID();
 
   public StreamMode getMode();
+
+  public void setBinaryXMPPExtensionReceiver(IBinaryXMPPExtensionReceiver receiver);
 }

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

@@ -3,19 +3,12 @@
 /**
  * Listener interface used by IStreamService and IBytestreamConnection to notify about established
  * or changed connections and incoming XMPP extensions.
- *
- * @author jurke
  */
 public interface IByteStreamConnectionListener {
 
-  /**
-   * Gets called when a {@linkplain BinaryXMPPExtension} was received.
-   *
-   * @param extension
-   */
-  public void receive(final BinaryXMPPExtension extension);
-
-  public void connectionClosed(String connectionID, IByteStreamConnection connection);
+  public default void connectionClosed(String connectionID, IByteStreamConnection connection) {
+    // NOP;
+  }
 
   /**
    * Gets called when a connection change is detected. The {@linkplain IByteStreamConnection
@@ -27,6 +20,8 @@ public interface IByteStreamConnectionListener {
    * @param incomingRequest <code>true</code> if the connection was a result of a remote connect
    *     request, <code>false</code> if the connect request was initiated on the local side
    */
-  public void connectionChanged(
-      String connectionID, IByteStreamConnection connection, boolean incomingRequest);
+  public default void connectionChanged(
+      String connectionID, IByteStreamConnection connection, boolean incomingRequest) {
+    // NOP;
+  }
 }