@Override
  public ServerQueueStatus handshakeWithSubscriptionFeed(Socket sock, boolean isPrimary)
      throws IOException, AuthenticationRequiredException, AuthenticationFailedException,
      ServerRefusedConnectionException, ClassNotFoundException {
    try {
      final DataOutputStream dos = new DataOutputStream(sock.getOutputStream());
      final InputStream in = sock.getInputStream();
      final DataInputStream dis = new DataInputStream(in);
      final DistributedMember member = getIDForSocket(sock);
      if (!multiuserSecureMode) {
        credentials = getCredentials(member);
      }
      final CommunicationMode mode = isPrimary ? CommunicationMode.PrimaryServerToClient
          : CommunicationMode.SecondaryServerToClient;
      write(dos, dis, mode, REPLY_OK, 0, new ArrayList<>(), credentials, member, true);

      // Wait here for a reply before continuing. This ensures that the client
      // updater is registered with the server before continuing.
      final byte acceptanceCode = dis.readByte();
      if (acceptanceCode == (byte) 21 && !(sock instanceof SSLSocket)) {
        // This is likely the case of server setup with SSL and client not using
        // SSL
        throw new AuthenticationRequiredException(
            "Server expecting SSL connection");
      }

      final byte endpointType = dis.readByte();
      final int queueSize = dis.readInt();

      // Read the message (if any)
      readMessage(dis, dos, acceptanceCode, member);

      final Map<Integer, List<String>> instantiatorMap = DataSerializer.readHashMap(dis);
      for (final Map.Entry<Integer, List<String>> entry : instantiatorMap.entrySet()) {
        final Integer id = entry.getKey();
        final List<String> instantiatorArguments = entry.getValue();
        InternalInstantiator.register(instantiatorArguments.get(0), instantiatorArguments.get(1),
            id, false);
      }

      final Map<Integer, String> dataSerializersMap = DataSerializer.readHashMap(dis);
      for (final Map.Entry<Integer, String> entry : dataSerializersMap.entrySet()) {
        InternalDataSerializer.register(entry.getValue(), false, null, null, entry.getKey());
      }
      final Map<Integer, List<String>> dsToSupportedClassNames = DataSerializer.readHashMap(dis);
      InternalDataSerializer.updateSupportedClassesMap(dsToSupportedClassNames);

      // the server's ping interval is only sent to subscription feeds so we can't read it as
      // part of a "standard" server response along with the other status data.
      final int pingInterval = dis.readInt();
      return new ServerQueueStatus(endpointType, queueSize, member, pingInterval);

    } catch (IOException | ClassNotFoundException ex) {
      CancelCriterion stopper = system.getCancelCriterion();
      stopper.checkCancelInProgress(null);
      throw ex;
    }
  }
Used by client-side CacheClientUpdater to handshake with a server in order to receive messages
generated by subscriptions (register-interest, continuous query)
geode-core/src/main/java/org/apache/geode/cache/client/internal/ClientSideHandshakeImpl.java