Add subscription connection parameters as payload of init message

apollo-runtime/src/main/java/com/apollographql/apollo/ApolloClient.java

@@ -246,6 +246,7 @@ private <D extends Operation.Data, T, V extends Operation.Variables> RealApolloC
     final List<ApolloInterceptor> applicationInterceptors = new ArrayList<>();
     boolean sendOperationIdentifiers;
     Optional<SubscriptionTransport.Factory> subscriptionTransportFactory = Optional.absent();
+    Optional<Map<String, Object>> subscriptionConnectionParams = Optional.absent();
 
     Builder() {
     }
@@ -432,6 +433,19 @@ public Builder subscriptionTransportFactory(@Nonnull SubscriptionTransport.Facto
       return this;
     }
 
+    /**
+     * <p>Sets up subscription connection parameters to be sent to the server when connection is established with
+     * subscription server</p>
+     *
+     * @param subscriptionConnectionParams map of connection parameters to be sent
+     * @return The {@link Builder} object to be used for chaining method calls
+     */
+    public Builder subscriptionConnectionParams(@Nonnull Map<String, Object> subscriptionConnectionParams) {
+      this.subscriptionConnectionParams = Optional.of(checkNotNull(subscriptionConnectionParams,
+          "subscriptionConnectionParams is null"));
+      return this;
+    }
+
     /**
      * Builds the {@link ApolloClient} instance using the configured values.
      *
@@ -474,7 +488,7 @@ public ApolloClient build() {
       Optional<SubscriptionTransport.Factory> subscriptionTransportFactory = this.subscriptionTransportFactory;
       if (subscriptionTransportFactory.isPresent()) {
         subscriptionManager = new RealSubscriptionManager(scalarTypeAdapters, subscriptionTransportFactory.get(),
-            dispatcher);
+            subscriptionConnectionParams.or(Collections.<String, Object>emptyMap()), dispatcher);
       }
 
       return new ApolloClient(serverUrl,

apollo-runtime/src/main/java/com/apollographql/apollo/internal/json/Utils.java

@@ -24,7 +24,7 @@ public static String toJsonString(@Nonnull Object data) throws IOException {
   }
 
   @SuppressWarnings("unchecked")
-  private static void writeToJson(Object value, JsonWriter jsonWriter) throws IOException {
+  public static void writeToJson(Object value, JsonWriter jsonWriter) throws IOException {
     if (value == null) {
       jsonWriter.nullValue();
     } else if (value instanceof Map) {

apollo-runtime/src/main/java/com/apollographql/apollo/internal/subscription/RealSubscriptionManager.java

@@ -35,6 +35,7 @@ public final class RealSubscriptionManager implements SubscriptionManager {
 
   private final ScalarTypeAdapters scalarTypeAdapters;
   private final SubscriptionTransport transport;
+  private Map<String, Object> connectionParams;
   private final Executor dispatcher;
   private final ResponseFieldMapperFactory responseFieldMapperFactory = new ResponseFieldMapperFactory();
   private final Runnable connectionAcknowledgeTimeoutTimerTask = new Runnable() {
@@ -49,12 +50,14 @@ public final class RealSubscriptionManager implements SubscriptionManager {
   };
 
   public RealSubscriptionManager(@Nonnull ScalarTypeAdapters scalarTypeAdapters,
-      @Nonnull final SubscriptionTransport.Factory transportFactory, @Nonnull final Executor dispatcher) {
+      @Nonnull final SubscriptionTransport.Factory transportFactory, @Nonnull Map<String, Object> connectionParams,
+      @Nonnull final Executor dispatcher) {
     checkNotNull(scalarTypeAdapters, "scalarTypeAdapters == null");
     checkNotNull(transportFactory, "transportFactory == null");
     checkNotNull(dispatcher, "dispatcher == null");
 
     this.scalarTypeAdapters = checkNotNull(scalarTypeAdapters, "scalarTypeAdapters == null");
+    this.connectionParams = checkNotNull(connectionParams, "connectionParams == null");
     this.transport = transportFactory.create(new SubscriptionTransportCallback(this, dispatcher));
     this.dispatcher = dispatcher;
   }
@@ -122,7 +125,7 @@ void doUnsubscribe(Subscription subscription) {
   void onTransportConnected() {
     synchronized (this) {
       state = State.CONNECTED;
-      transport.send(new OperationClientMessage.Init());
+      transport.send(new OperationClientMessage.Init(connectionParams));
     }
 
     timer.schedule(CONNECTION_ACKNOWLEDGE_TIMEOUT_TIMER_TASK_ID, connectionAcknowledgeTimeoutTimerTask,

apollo-runtime/src/main/java/com/apollographql/apollo/subscription/OperationClientMessage.java

@@ -3,9 +3,11 @@
 import com.apollographql.apollo.api.Subscription;
 import com.apollographql.apollo.internal.json.InputFieldJsonWriter;
 import com.apollographql.apollo.internal.json.JsonWriter;
+import com.apollographql.apollo.internal.json.Utils;
 import com.apollographql.apollo.response.ScalarTypeAdapters;
 
 import java.io.IOException;
+import java.util.Map;
 
 import javax.annotation.Nonnull;
 
@@ -40,10 +42,19 @@ public String toJsonString() {
 
   public static final class Init extends OperationClientMessage {
     private static final String TYPE = "connection_init";
+    private final Map<String, Object> connectionParams;
+
+    public Init(@Nonnull Map<String, Object> connectionParams) {
+      this.connectionParams = checkNotNull(connectionParams, "connectionParams == null");
+    }
 
     @Override public void writeToJson(@Nonnull JsonWriter writer) throws IOException {
       checkNotNull(writer, "writer == null");
       writer.name(JSON_KEY_TYPE).value(TYPE);
+      if (!connectionParams.isEmpty()) {
+        writer.name(JSON_KEY_PAYLOAD);
+        Utils.writeToJson(connectionParams, writer);
+      }
     }
   }
 

apollo-runtime/src/test/java/com/apollographql/apollo/internal/subscription/SubscriptionManagerTest.java

@@ -37,7 +37,8 @@ public class SubscriptionManagerTest {
 
   @Before public void setUp() throws Exception {
     subscriptionTransportFactory = new MockSubscriptionTransportFactory();
-    subscriptionManager = new RealSubscriptionManager(new ScalarTypeAdapters(Collections.<ScalarType, CustomTypeAdapter>emptyMap()), subscriptionTransportFactory, new MockExecutor());
+    subscriptionManager = new RealSubscriptionManager(new ScalarTypeAdapters(Collections.<ScalarType, CustomTypeAdapter>emptyMap()),
+        subscriptionTransportFactory, Collections.<String, Object>emptyMap(), new MockExecutor());
     assertThat(subscriptionTransportFactory.subscriptionTransport).isNotNull();
     assertThat(subscriptionManager.state).isEqualTo(RealSubscriptionManager.State.DISCONNECTED);
   }

apollo-runtime/src/test/java/com/apollographql/apollo/subscription/WebSocketSubscriptionTransportMessageTest.java

@@ -44,8 +44,18 @@ public class WebSocketSubscriptionTransportMessageTest {
   }
 
   @Test public void connectionInit() {
-    subscriptionTransport.send(new OperationClientMessage.Init());
+    subscriptionTransport.send(new OperationClientMessage.Init(Collections.<String, Object>emptyMap()));
     assertThat(webSocketFactory.webSocket.lastSentMessage).isEqualTo("{\"type\":\"connection_init\"}");
+
+    subscriptionTransport.send(
+        new OperationClientMessage.Init(
+            new UnmodifiableMapBuilder<String, Object>()
+                .put("param1", true)
+                .put("param2", "value")
+                .build()
+        )
+    );
+    assertThat(webSocketFactory.webSocket.lastSentMessage).isEqualTo("{\"type\":\"connection_init\",\"payload\":{\"param1\":true,\"param2\":\"value\"}}");
   }
 
   @Test public void startSubscription() {

apollo-runtime/src/test/java/com/apollographql/apollo/subscription/WebSocketSubscriptionTransportTest.java

@@ -3,6 +3,7 @@
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.Collections;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -63,18 +64,18 @@ public class WebSocketSubscriptionTransportTest {
 
   @Test public void send() {
     try {
-      subscriptionTransport.send(new OperationClientMessage.Init());
+      subscriptionTransport.send(new OperationClientMessage.Init(Collections.<String, Object>emptyMap()));
       fail("expected IllegalStateException");
     } catch (IllegalStateException expected) {
       // expected
     }
 
     subscriptionTransport.connect();
-    subscriptionTransport.send(new OperationClientMessage.Init());
+    subscriptionTransport.send(new OperationClientMessage.Init(Collections.<String, Object>emptyMap()));
     subscriptionTransport.disconnect(new OperationClientMessage.Terminate());
 
     try {
-      subscriptionTransport.send(new OperationClientMessage.Init());
+      subscriptionTransport.send(new OperationClientMessage.Init(Collections.<String, Object>emptyMap()));
       fail("expected IllegalStateException");
     } catch (IllegalStateException expected) {
       // expected