StreamingConnectionImpl connect ( ) throws IOException , InterruptedException { boolean exThrown = false ; io . nats . client . Connection nc = getNatsConnection ( ) ; // Create a NATS connection if it doesn't exist if ( nc == null ) { nc = createNatsConnection ( ) ; setNatsConnection ( nc ) ; ncOwned = true ; } else if ( nc . getStatus ( ) != Connection . Status . CONNECTED ) { // Bail if the custom NATS connection is disconnected throw new IOException ( NatsStreaming . ERR_BAD_CONNECTION ) ; } try { this . hbSubject = this . newInbox ( ) ; this . ackSubject = String . format ( "%s.%s" , NatsStreaming . DEFAULT_ACK_PREFIX , this . nuid . next ( ) ) ; this . ackDispatcher = nc . createDispatcher ( msg -> { this . processAck ( msg ) ; } ) ; this . heartbeatDispatcher = nc . createDispatcher ( msg -> { this . processHeartBeat ( msg ) ; } ) ; this . messageDispatcher = nc . createDispatcher ( msg -> { this . processMsg ( msg ) ; } ) ; this . heartbeatDispatcher . subscribe ( this . hbSubject ) ; this . ackDispatcher . subscribe ( this . ackSubject ) ; this . heartbeatDispatcher . setPendingLimits ( - 1 , - 1 ) ; this . ackDispatcher . setPendingLimits ( - 1 , - 1 ) ; this . messageDispatcher . setPendingLimits ( - 1 , - 1 ) ; this . customDispatchers = new HashMap <> ( ) ; // Send Request to discover the cluster String discoverSubject = String . format ( "%s.%s" , opts . getDiscoverPrefix ( ) , clusterId ) ; ConnectRequest req = ConnectRequest . newBuilder ( ) . setClientID ( clientId ) . setHeartbeatInbox ( this . hbSubject ) . build ( ) ; byte [ ] bytes = req . toByteArray ( ) ; Message reply = nc . request ( discoverSubject , bytes , opts . getConnectTimeout ( ) ) ; if ( reply == null ) { throw new IOException ( ERR_CONNECTION_REQ_TIMEOUT ) ; } ConnectResponse cr = ConnectResponse . parseFrom ( reply . getData ( ) ) ; if ( ! cr . getError ( ) . isEmpty ( ) ) { // This is already a properly formatted streaming error message // (StreamingConnectionImpl.SERVER_ERR_INVALID_CLIENT) throw new IOException ( cr . getError ( ) ) ; } // Capture cluster configuration endpoints to publish and // subscribe/unsubscribe. pubPrefix = cr . getPubPrefix ( ) ; subRequests = cr . getSubRequests ( ) ; unsubRequests = cr . getUnsubRequests ( ) ; subCloseRequests = cr . getSubCloseRequests ( ) ; closeRequests = cr . getCloseRequests ( ) ; // Setup the ACK subscription pubAckMap = new HashMap <> ( ) ; // Create Subscription map subMap = new HashMap <> ( ) ; pubAckChan = new LinkedBlockingQueue <> ( opts . getMaxPubAcksInFlight ( ) ) ; } catch ( IOException e ) { exThrown = true ; throw e ; } finally { if ( exThrown ) { try { close ( ) ; } catch ( Exception e ) { /* NOOP -- can't do anything if close fails */ } } } return this ; }
Connect will form a connection to the STAN subsystem .