Share MessageRetrievalService's pipe by reference.

java/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessageSender.java

@@ -81,11 +81,11 @@ public class SignalServiceMessageSender {
 
   private static final String TAG = SignalServiceMessageSender.class.getSimpleName();
 
-  private final PushServiceSocket                                   socket;
-  private final SignalProtocolStore                                 store;
-  private final SignalServiceAddress                                localAddress;
-  private final AtomicReference<Optional<SignalServiceMessagePipe>> pipe;
-  private final Optional<EventListener>                             eventListener;
+  private final PushServiceSocket                         socket;
+  private final SignalProtocolStore                       store;
+  private final SignalServiceAddress                      localAddress;
+  private final AtomicReference<SignalServiceMessagePipe> pipe;
+  private final Optional<EventListener>                   eventListener;
 
   /**
    * Construct a SignalServiceMessageSender.
@@ -101,7 +101,7 @@ public SignalServiceMessageSender(SignalServiceConfiguration urls,
                                     String user, String password,
                                     SignalProtocolStore store,
                                     String userAgent,
-                                    Optional<SignalServiceMessagePipe> pipe,
+                                    AtomicReference<SignalServiceMessagePipe> pipe,
                                     Optional<EventListener> eventListener)
   {
     this(urls, new StaticCredentialsProvider(user, password, null), store, userAgent, pipe, eventListener);
@@ -111,13 +111,13 @@ public SignalServiceMessageSender(SignalServiceConfiguration urls,
                                     CredentialsProvider credentialsProvider,
                                     SignalProtocolStore store,
                                     String userAgent,
-                                    Optional<SignalServiceMessagePipe> pipe,
+                                    AtomicReference<SignalServiceMessagePipe> pipe,
                                     Optional<EventListener> eventListener)
   {
     this.socket        = new PushServiceSocket(urls, credentialsProvider, userAgent);
     this.store         = store;
     this.localAddress  = new SignalServiceAddress(credentialsProvider.getUser());
-    this.pipe          = new AtomicReference<>(pipe);
+    this.pipe          = pipe;
     this.eventListener = eventListener;
   }
 
@@ -243,10 +243,6 @@ public void cancelInFlightRequests() {
     socket.cancelInFlightRequests();
   }
 
-  public void setMessagePipe(SignalServiceMessagePipe pipe) {
-    this.pipe.set(Optional.fromNullable(pipe));
-  }
-
   private void sendMessage(VerifiedMessage message) throws IOException, UntrustedIdentityException {
     byte[] nullMessageBody = DataMessage.newBuilder()
                                         .setBody(Base64.encodeBytes(Util.getRandomLengthBytes(140)))
@@ -652,12 +648,12 @@ private SendMessageResponse sendMessage(SignalServiceAddress recipient, long tim
     for (int i=0;i<3;i++) {
       try {
         OutgoingPushMessageList            messages = getEncryptedMessages(socket, recipient, timestamp, content, silent);
-        Optional<SignalServiceMessagePipe> pipe     = this.pipe.get();
+        SignalServiceMessagePipe pipe               = this.pipe.get();
 
-        if (pipe.isPresent()) {
+        if (pipe != null) {
           try {
             Log.w(TAG, "Transmitting over pipe...");
-            return pipe.get().send(messages);
+            return pipe.send(messages);
           } catch (IOException e) {
             Log.w(TAG, e);
             Log.w(TAG, "Falling back to new connection...");