ConnectionManager.java

package com.ubercab.network.ramengrpc;

import com.uber.streaming.ramen.RamenStreamingResponse;
import io.grpc.stub.StreamObserver;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class ConnectionManager {

  private ConnectionState curConnectionState = ConnectionState.Disconected;
  private LinkedBlockingQueue<EventInfo> eventQueue = new LinkedBlockingQueue();
  private StreamObserver<RamenStreamingResponse> grpcResponseStream;
  private boolean receivedInitRamen = false;
  private boolean lifeCycleStartReceived = false;
  private long latestFailoverReconnectEventTimeMs = -1;
  private long connectionInitiationTimeMs = -1;

  public ConnectionManager() {

  }

  public void run() throws InterruptedException {
    EventInfo eventInfo = eventQueue.poll(5, TimeUnit.SECONDS);
    while (eventInfo != null) {
      // process events.
      synchronized (this) {
        // if cur state is disconnected, process only new connect events.
        if (curConnectionState == ConnectionState.Disconected) {
          if (eventInfo.getEvent() == ConnectDisconnectEvent.INITIALIZE_RAMEN
              || eventInfo.getEvent() == ConnectDisconnectEvent.LIFE_CYCLE_START) {
            if (eventInfo.getEvent() == ConnectDisconnectEvent.INITIALIZE_RAMEN) {
              receivedInitRamen = true;
            }
            if (eventInfo.getEvent() == ConnectDisconnectEvent.LIFE_CYCLE_START) {
              lifeCycleStartReceived = true;
            }
            if (receivedInitRamen && lifeCycleStartReceived) {
              curConnectionState = ConnectionState.ReconnectionInProgress;
              connect();
            }
          }
        } else if(curConnectionState == ConnectionState.ReconnectionInProgress) {
          // if(eventInfo.getEvent() is an failover reconnect event
          latestFailoverReconnectEventTimeMs = System.currentTimeMillis();
          // no op.
        } else { // irrespective of whether disconnect state or connected state.
          // initiate a reconnection.
                // disconnect if required
                // start a new call i.e. connection initiation.
                // waits for first resp msg to update the connection status
        }
      }
      eventInfo = eventQueue.poll(5, TimeUnit.SECONDS);
    }
  }

  private void connect(){
    // start to establish a connection.
    connectionInitiationTimeMs = System.currentTimeMillis();
    // proceed on to sending a new connection.
  }

  private void initResponseStreamHandler() {
    this.grpcResponseStream =
        new StreamObserver<RamenStreamingResponse>() {
          @Override
          public void onNext(RamenStreamingResponse response) {
            // if first response msg, then move state to ConnectionState.Connect and inform
            // GrpcService.
            // relay the msg to grpc service.
            synchronized (this) {
              if (curConnectionState == ConnectionState.ReconnectionInProgress) {
                // latestReconnectEventTimeMs == -1  no op.
                if (latestFailoverReconnectEventTimeMs > connectionInitiationTimeMs) {
                  try {
                    eventQueue
                        .put(new EventInfo(ConnectDisconnectEvent.FAILOVER, System.currentTimeMillis()));
                  } catch (InterruptedException e) {
                    e.printStackTrace();
                  }
                } else {
                  curConnectionState = ConnectionState.Connected;
                  // inform grpc service of connection establishment
                  // relay msg.
                }
              } else {
                // relay msg to grpc service.
              }
            }
          }

          @Override
          public void onError(Throwable throwable) {
            // if cur state is connected, disconnect
            // start reconnection.
            try {
              eventQueue.put(new EventInfo(ConnectDisconnectEvent.STREAM_ERROR, System.currentTimeMillis()));
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
          }

          @Override
          public void onCompleted() {
            // if cur state is connected, disconnect
            // start reconnection
            try {
              eventQueue.put(new EventInfo(ConnectDisconnectEvent.STREAM_CLOSE, System.currentTimeMillis()));
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
          }
        };
  }

  public void updateConnectionManagement(ConnectDisconnectEvent event) throws InterruptedException {
    eventQueue.put(new EventInfo(event, System.currentTimeMillis()));
  }

  class EventInfo {

    private ConnectDisconnectEvent event;
    private long eventTimeMs;

    public EventInfo(ConnectDisconnectEvent event, long eventTimeMs) {
      this.event = event;
      this.eventTimeMs = eventTimeMs;
    }

    public ConnectDisconnectEvent getEvent() {
      return event;
    }

    public long getEventTimeMs() {
      return eventTimeMs;
    }
  }


  enum ConnectionState {
    INITIATED,
    Connected,
    Disconected,
    ReconnectionInProgress
  }

  enum ConnectDisconnectEvent {
    /**
     * Failover connection update event is received when there is a need to disconnect and reconnect
     * to a new host.
     */
    FAILOVER,

    /**
     * HEARTBEAT_TIMEOUT connection update event is sent by the RamenClient when there has been no
     * message received by the client within a certain threshold. In this case, we try to close the
     * connection, and reconnect again.
     */
    HEARTBEAT_TIMEOUT,

    /**
     * INITIALIZE_RAMEN connection update event signifies that all Ramen consumers have been
     * registered and we are now ready to connect to Ramen and start receiving messages which can be
     * consumed by RamenConsumers. In this case, there should not be any previous connection present
     * already and only requires a connect to happen.
     */
    INITIALIZE_RAMEN,

    /**
     * LIFE_CYCLE_START connection update event marks that the Ramen client should be started now.
     * This would typically happen when the app goes into the foreground but may differ from app to
     * app. In such case, we do not expect a connection to be present already, and we can simply
     * connect.
     */
    LIFE_CYCLE_START,

    /**
     * LIFE_CYCLE_STOP connection update event triggers that the Ramen life cycle has stopped and
     * the connection should be closed now. This would typically happen when the app goes into the
     * background but may differ from app to app. In such case, we only want to disconnect and not
     * connect again.
     */
    LIFE_CYCLE_STOP,

    /**
     * NETWORK_AVAILABLE connection update event signifies that the app is now connected to a
     * network, and requires the client to connect again. Since network was not available
     * previously, the app is expected to be in a disconnected state already.
     */
    NETWORK_AVAILABLE,

    /**
     * NETWORK_UNAVAILABLE connection update event signifies that network connection is not present
     * and therefore, the connection would be disconnected. It does not try to connect again.
     */
    NETWORK_UNAVAILABLE,

    /**
     * SERVER_CLOSE connection update event signifies that the server has asked the client to
     * disconnect and connect again.
     */
    SERVER_CLOSE,

    /**
     * STREAM_CLOSE connection update event signifies that the stream has been closed unexpectedly
     * by the server and therefore the client should connect again.
     */
    STREAM_CLOSE,

    /**
     * STREAM_ERROR connection update event signifies that the stream has received an unexpected
     * error on the response stream and therefore the client should connect again.
     */
    STREAM_ERROR,
    /**
     * UNAUTHORIZED connection update event signifies that the connection could not be created due
     * to an authentication error. The client should try to connect again.
     */
    UNAUTHORIZED
  }
}