geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/HandShake.java
public ServerQueueStatus handshakeWithServer(Connection conn, ServerLocation location,
      CommunicationMode communicationMode) throws IOException, AuthenticationRequiredException,
      AuthenticationFailedException, ServerRefusedConnectionException {
    try {
      ServerQueueStatus serverQStatus = null;
      Socket sock = conn.getSocket();
      DataOutputStream dos = new DataOutputStream(sock.getOutputStream());
      final InputStream in = sock.getInputStream();
      DataInputStream dis = new DataInputStream(in);
      DistributedMember member = getIDForSocket(sock);
      // if running in a loner system, use the new port number in the ID to
      // help differentiate from other clients
      DM dm = ((InternalDistributedSystem) this.system).getDistributionManager();
      InternalDistributedMember idm = dm.getDistributionManagerId();
      synchronized (idm) {
        if (idm.getPort() == 0 && dm instanceof LonerDistributionManager) {
          int port = sock.getLocalPort();
          ((LonerDistributionManager) dm).updateLonerPort(port);
          updateProxyID(dm.getDistributionManagerId());
        }
      }
      if (communicationMode.isWAN()) {
        this.credentials = getCredentials(member);
      }
      byte intermediateAcceptanceCode = write(dos, dis, communicationMode, REPLY_OK,
          this.clientReadTimeout, null, this.credentials, member, false);

      String authInit = this.system.getProperties().getProperty(SECURITY_CLIENT_AUTH_INIT);
      if (!communicationMode.isWAN() && intermediateAcceptanceCode != REPLY_AUTH_NOT_REQUIRED
          && (authInit != null && authInit.length() != 0)) {
        location.compareAndSetRequiresCredentials(true);
      }
      // Read the acceptance code
      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(
            LocalizedStrings.HandShake_SERVER_EXPECTING_SSL_CONNECTION.toLocalizedString());
      }
      if (acceptanceCode == REPLY_SERVER_IS_LOCATOR) {
        throw new GemFireConfigException("Improperly configured client detected.  " + "Server at "
            + location + " is actually a locator.  Use addPoolLocator to configure locators.");
      }

      // Successful handshake for GATEWAY_TO_GATEWAY mode sets the peer version in connection
      if (communicationMode.isWAN() && !(acceptanceCode == REPLY_EXCEPTION_AUTHENTICATION_REQUIRED
          || acceptanceCode == REPLY_EXCEPTION_AUTHENTICATION_FAILED)) {
        short wanSiteVersion = Version.readOrdinal(dis);
        conn.setWanSiteVersion(wanSiteVersion);
        // establish a versioned stream for the other site, if necessary
        if (wanSiteVersion < Version.CURRENT_ORDINAL) {
          dis = new VersionedDataInputStream(dis, Version.fromOrdinalOrCurrent(wanSiteVersion));
        }
      }

      // No need to check for return value since DataInputStream already throws
      // EOFException in case of EOF
      byte epType = dis.readByte();
      int qSize = dis.readInt();

      // Read the server member
      member = readServerMember(dis);
      serverQStatus = new ServerQueueStatus(epType, qSize, member);

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

      // Read delta-propagation property value from server.
      // [sumedh] Static variable below? Client can connect to different
      // DSes with different values of this. It shoule be a member variable.
      if (!communicationMode.isWAN() && currentClientVersion.compareTo(Version.GFE_61) >= 0) {
        deltaEnabledOnServer = dis.readBoolean();
      }

      // validate that the remote side has a different distributed system id.
      if (communicationMode.isWAN() && Version.GFE_66.compareTo(conn.getWanSiteVersion()) <= 0
          && currentClientVersion.compareTo(Version.GFE_66) >= 0) {
        int remoteDistributedSystemId = in.read();
        int localDistributedSystemId =
            ((InternalDistributedSystem) system).getDistributionManager().getDistributedSystemId();
        if (localDistributedSystemId >= 0
            && localDistributedSystemId == remoteDistributedSystemId) {
          throw new GatewayConfigurationException(
              "Remote WAN site's distributed system id " + remoteDistributedSystemId
                  + " matches this sites distributed system id " + localDistributedSystemId);
        }
      }
      // Read the PDX registry size from the remote size
      if (communicationMode.isWAN() && Version.GFE_80.compareTo(conn.getWanSiteVersion()) <= 0
          && currentClientVersion.compareTo(Version.GFE_80) >= 0) {
        int remotePdxSize = dis.readInt();
        serverQStatus.setPdxSize(remotePdxSize);
      }

      return serverQStatus;
    } catch (IOException ex) {
      CancelCriterion stopper = this.system.getCancelCriterion();
      stopper.checkCancelInProgress(null);
      throw ex;
    }
  }
/**
   * Client-side handshake with a Server
   */
Client-side handshake with a Server
Client - side handshake with a Server
public ServerQueueStatus handshakeWithServer(Connection conn, ServerLocation location,
      CommunicationMode communicationMode) throws IOException, AuthenticationRequiredException,
      AuthenticationFailedException, ServerRefusedConnectionException {
    try {
      ServerQueueStatus serverQStatus = null;
      Socket sock = conn.getSocket();
      DataOutputStream dos = new DataOutputStream(sock.getOutputStream());
      final InputStream in = sock.getInputStream();
      DataInputStream dis = new DataInputStream(in);
      DistributedMember member = getIDForSocket(sock);
      DM dm = ((InternalDistributedSystem) this.system).getDistributionManager();
      InternalDistributedMember idm = dm.getDistributionManagerId();
      synchronized (idm) {
        if (idm.getPort() == 0 && dm instanceof LonerDistributionManager) {
          int port = sock.getLocalPort();
          ((LonerDistributionManager) dm).updateLonerPort(port);
          updateProxyID(dm.getDistributionManagerId());
        }
      }
      if (communicationMode.isWAN()) {
        this.credentials = getCredentials(member);
      }
      byte intermediateAcceptanceCode = write(dos, dis, communicationMode, REPLY_OK,
          this.clientReadTimeout, null, this.credentials, member, false);
      String authInit = this.system.getProperties().getProperty(SECURITY_CLIENT_AUTH_INIT);
      if (!communicationMode.isWAN() && intermediateAcceptanceCode != REPLY_AUTH_NOT_REQUIRED
          && (authInit != null && authInit.length() != 0)) {
        location.compareAndSetRequiresCredentials(true);
      }
      byte acceptanceCode = dis.readByte();
      if (acceptanceCode == (byte) 21 && !(sock instanceof SSLSocket)) {
        throw new AuthenticationRequiredException(
            LocalizedStrings.HandShake_SERVER_EXPECTING_SSL_CONNECTION.toLocalizedString());
      }
      if (acceptanceCode == REPLY_SERVER_IS_LOCATOR) {
        throw new GemFireConfigException("Improperly configured client detected.  " + "Server at "
            + location + " is actually a locator.  Use addPoolLocator to configure locators.");
      }
      if (communicationMode.isWAN() && !(acceptanceCode == REPLY_EXCEPTION_AUTHENTICATION_REQUIRED
          || acceptanceCode == REPLY_EXCEPTION_AUTHENTICATION_FAILED)) {
        short wanSiteVersion = Version.readOrdinal(dis);
        conn.setWanSiteVersion(wanSiteVersion);
        if (wanSiteVersion < Version.CURRENT_ORDINAL) {
          dis = new VersionedDataInputStream(dis, Version.fromOrdinalOrCurrent(wanSiteVersion));
        }
      }
      byte epType = dis.readByte();
      int qSize = dis.readInt();
      member = readServerMember(dis);
      serverQStatus = new ServerQueueStatus(epType, qSize, member);
      readMessage(dis, dos, acceptanceCode, member);
      if (!communicationMode.isWAN() && currentClientVersion.compareTo(Version.GFE_61) >= 0) {
        deltaEnabledOnServer = dis.readBoolean();
      }
      if (communicationMode.isWAN() && Version.GFE_66.compareTo(conn.getWanSiteVersion()) <= 0
          && currentClientVersion.compareTo(Version.GFE_66) >= 0) {
        int remoteDistributedSystemId = in.read();
        int localDistributedSystemId =
            ((InternalDistributedSystem) system).getDistributionManager().getDistributedSystemId();
        if (localDistributedSystemId >= 0
            && localDistributedSystemId == remoteDistributedSystemId) {
          throw new GatewayConfigurationException(
              "Remote WAN site's distributed system id " + remoteDistributedSystemId
                  + " matches this sites distributed system id " + localDistributedSystemId);
        }
      }
      if (communicationMode.isWAN() && Version.GFE_80.compareTo(conn.getWanSiteVersion()) <= 0
          && currentClientVersion.compareTo(Version.GFE_80) >= 0) {
        int remotePdxSize = dis.readInt();
        serverQStatus.setPdxSize(remotePdxSize);
      }
      return serverQStatus;
    } catch (IOException ex) {
      CancelCriterion stopper = this.system.getCancelCriterion();
      stopper.checkCancelInProgress(null);
      throw ex;
    }
  }
public ServerQueueStatus handshakeWithServer ( Connection conn , ServerLocation location , CommunicationMode communicationMode ) throws IOException , AuthenticationRequiredException , AuthenticationFailedException , ServerRefusedConnectionException { try { ServerQueueStatus serverQStatus = null ; Socket sock = conn . getSocket ( ) ; DataOutputStream dos = new DataOutputStream ( sock . getOutputStream ( ) ) ; final InputStream in = sock . getInputStream ( ) ; DataInputStream dis = new DataInputStream ( in ) ; DistributedMember member = getIDForSocket ( sock ) ; DM dm = ( ( InternalDistributedSystem ) this . system ) . getDistributionManager ( ) ; InternalDistributedMember idm = dm . getDistributionManagerId ( ) ; synchronized ( idm ) { if ( idm . getPort ( ) == 0 && dm instanceof LonerDistributionManager ) { int port = sock . getLocalPort ( ) ; ( ( LonerDistributionManager ) dm ) . updateLonerPort ( port ) ; updateProxyID ( dm . getDistributionManagerId ( ) ) ; } } if ( communicationMode . isWAN ( ) ) { this . credentials = getCredentials ( member ) ; } byte intermediateAcceptanceCode = write ( dos , dis , communicationMode , REPLY_OK , this . clientReadTimeout , null , this . credentials , member , false ) ; String authInit = this . system . getProperties ( ) . getProperty ( SECURITY_CLIENT_AUTH_INIT ) ; if ( ! communicationMode . isWAN ( ) && intermediateAcceptanceCode != REPLY_AUTH_NOT_REQUIRED && ( authInit != null && authInit . length ( ) != 0 ) ) { location . compareAndSetRequiresCredentials ( true ) ; } byte acceptanceCode = dis . readByte ( ) ; if ( acceptanceCode == ( byte ) 21 && ! ( sock instanceof SSLSocket ) ) { throw new AuthenticationRequiredException ( LocalizedStrings . HandShake_SERVER_EXPECTING_SSL_CONNECTION . toLocalizedString ( ) ) ; } if ( acceptanceCode == REPLY_SERVER_IS_LOCATOR ) { throw new GemFireConfigException ( "Improperly configured client detected.  " + "Server at " + location + " is actually a locator.  Use addPoolLocator to configure locators." ) ; } if ( communicationMode . isWAN ( ) && ! ( acceptanceCode == REPLY_EXCEPTION_AUTHENTICATION_REQUIRED || acceptanceCode == REPLY_EXCEPTION_AUTHENTICATION_FAILED ) ) { short wanSiteVersion = Version . readOrdinal ( dis ) ; conn . setWanSiteVersion ( wanSiteVersion ) ; if ( wanSiteVersion < Version . CURRENT_ORDINAL ) { dis = new VersionedDataInputStream ( dis , Version . fromOrdinalOrCurrent ( wanSiteVersion ) ) ; } } byte epType = dis . readByte ( ) ; int qSize = dis . readInt ( ) ; member = readServerMember ( dis ) ; serverQStatus = new ServerQueueStatus ( epType , qSize , member ) ; readMessage ( dis , dos , acceptanceCode , member ) ; if ( ! communicationMode . isWAN ( ) && currentClientVersion . compareTo ( Version . GFE_61 ) >= 0 ) { deltaEnabledOnServer = dis . readBoolean ( ) ; } if ( communicationMode . isWAN ( ) && Version . GFE_66 . compareTo ( conn . getWanSiteVersion ( ) ) <= 0 && currentClientVersion . compareTo ( Version . GFE_66 ) >= 0 ) { int remoteDistributedSystemId = in . read ( ) ; int localDistributedSystemId = ( ( InternalDistributedSystem ) system ) . getDistributionManager ( ) . getDistributedSystemId ( ) ; if ( localDistributedSystemId >= 0 && localDistributedSystemId == remoteDistributedSystemId ) { throw new GatewayConfigurationException ( "Remote WAN site's distributed system id " + remoteDistributedSystemId + " matches this sites distributed system id " + localDistributedSystemId ) ; } } if ( communicationMode . isWAN ( ) && Version . GFE_80 . compareTo ( conn . getWanSiteVersion ( ) ) <= 0 && currentClientVersion . compareTo ( Version . GFE_80 ) >= 0 ) { int remotePdxSize = dis . readInt ( ) ; serverQStatus . setPdxSize ( remotePdxSize ) ; } return serverQStatus ; } catch ( IOException ex ) { CancelCriterion stopper = this . system . getCancelCriterion ( ) ; stopper . checkCancelInProgress ( null ) ; throw ex ; } }
Client-side handshake with a Server
Client - side handshake with a Server
// if running in a loner system, use the new port number in the ID to // help differentiate from other clients // Read the acceptance code // This is likely the case of server setup with SSL and client not using // SSL // Successful handshake for GATEWAY_TO_GATEWAY mode sets the peer version in connection // establish a versioned stream for the other site, if necessary // No need to check for return value since DataInputStream already throws // EOFException in case of EOF // Read the server member // Read the message (if any) // Read delta-propagation property value from server. // [sumedh] Static variable below? Client can connect to different // DSes with different values of this. It shoule be a member variable. // validate that the remote side has a different distributed system id. // Read the PDX registry size from the remote size