agent/acs/handler/acs_handler.go
func (acsSession *session) Start() error {
	// connectToACS channel is used to indicate the intent to connect to ACS
	// It's processed by the select loop to connect to ACS
	connectToACS := make(chan struct{})
	// This is required to trigger the first connection to ACS. Subsequent
	// connections are triggered by the handleACSError() method
	go func() {
		connectToACS <- struct{}{}
	}()
	for {
		select {
		case <-connectToACS:
			seelog.Debugf("Received connect to ACS message")
			// Start a session with ACS
			acsError := acsSession.startSessionOnce()
			// Session with ACS was stopped with some error, start processing the error
			isInactiveInstance := isInactiveInstanceError(acsError)
			if isInactiveInstance {
				// If the instance was deregistered, send an event to the event stream
				// for the same
				seelog.Debug("Container instance is deregistered, notifying listeners")
				err := acsSession.deregisterInstanceEventStream.WriteToEventStream(struct{}{})
				if err != nil {
					seelog.Debugf("Failed to write to deregister container instance event stream, err: %v", err)
				}
			}
			if shouldReconnectWithoutBackoff(acsError) {
				// If ACS closed the connection, there's no need to backoff,
				// reconnect immediately
				seelog.Info("ACS Websocket connection closed for a valid reason")
				acsSession.backoff.Reset()
				sendEmptyMessageOnChannel(connectToACS)
			} else {
				// Disconnected unexpectedly from ACS, compute backoff duration to
				// reconnect
				reconnectDelay := acsSession.computeReconnectDelay(isInactiveInstance)
				seelog.Infof("Reconnecting to ACS in: %s", reconnectDelay.String())
				waitComplete := acsSession.waitForDuration(reconnectDelay)
				if waitComplete {
					// If the context was not cancelled and we've waited for the
					// wait duration without any errors, send the message to the channel
					// to reconnect to ACS
					sendEmptyMessageOnChannel(connectToACS)
				} else {
					// Wait was interrupted. We expect the session to close as canceling
					// the session context is the only way to end up here. Print a message
					// to indicate the same
					seelog.Info("Interrupted waiting for reconnect delay to elapse; Expect session to close")
				}
			}
		case <-acsSession.ctx.Done():
			seelog.Debugf("ACS session context cancelled")
			return acsSession.ctx.Err()
		}

	}
}
func (acsSession *session) Start() error {
	// connectToACS channel is used to indicate the intent to connect to ACS
	// It's processed by the select loop to connect to ACS
	connectToACS := make(chan struct{})
	// This is required to trigger the first connection to ACS. Subsequent
	// connections are triggered by the handleACSError() method
	go func() {
		connectToACS <- struct{}{}
	}()
	for {
		select {
		case <-connectToACS:
			seelog.Debugf("Received connect to ACS message")
			// Start a session with ACS
			acsError := acsSession.startSessionOnce()
			// Session with ACS was stopped with some error, start processing the error
			isInactiveInstance := isInactiveInstanceError(acsError)
			if isInactiveInstance {
				// If the instance was deregistered, send an event to the event stream
				// for the same
				seelog.Debug("Container instance is deregistered, notifying listeners")
				err := acsSession.deregisterInstanceEventStream.WriteToEventStream(struct{}{})
				if err != nil {
					seelog.Debugf("Failed to write to deregister container instance event stream, err: %v", err)
				}
			}
			if shouldReconnectWithoutBackoff(acsError) {
				// If ACS closed the connection, there's no need to backoff,
				// reconnect immediately
				seelog.Info("ACS Websocket connection closed for a valid reason")
				acsSession.backoff.Reset()
				sendEmptyMessageOnChannel(connectToACS)
			} else {
				// Disconnected unexpectedly from ACS, compute backoff duration to
				// reconnect
				reconnectDelay := acsSession.computeReconnectDelay(isInactiveInstance)
				seelog.Infof("Reconnecting to ACS in: %s", reconnectDelay.String())
				waitComplete := acsSession.waitForDuration(reconnectDelay)
				if waitComplete {
					// If the context was not cancelled and we've waited for the
					// wait duration without any errors, send the message to the channel
					// to reconnect to ACS
					sendEmptyMessageOnChannel(connectToACS)
				} else {
					// Wait was interrupted. We expect the session to close as canceling
					// the session context is the only way to end up here. Print a message
					// to indicate the same
					seelog.Info("Interrupted waiting for reconnect delay to elapse; Expect session to close")
				}
			}
		case <-acsSession.ctx.Done():
			seelog.Debugf("ACS session context cancelled")
			return acsSession.ctx.Err()
		}

	}
}
func ( acsSession * session ) Start ( ) error { // connectToACS channel is used to indicate the intent to connect to ACS // It's processed by the select loop to connect to ACS connectToACS := make ( chan struct { } ) // This is required to trigger the first connection to ACS. Subsequent // connections are triggered by the handleACSError() method go func ( ) { connectToACS <- struct { } { } } ( ) for { select { case <- connectToACS : seelog . Debugf ( " " ) // Start a session with ACS acsError := acsSession . startSessionOnce ( ) // Session with ACS was stopped with some error, start processing the error isInactiveInstance := isInactiveInstanceError ( acsError ) if isInactiveInstance { // If the instance was deregistered, send an event to the event stream // for the same seelog . Debug ( " " ) err := acsSession . deregisterInstanceEventStream . WriteToEventStream ( struct { } { } ) if err != nil { seelog . Debugf ( " " , err ) } } if shouldReconnectWithoutBackoff ( acsError ) { // If ACS closed the connection, there's no need to backoff, // reconnect immediately seelog . Info ( " " ) acsSession . backoff . Reset ( ) sendEmptyMessageOnChannel ( connectToACS ) } else { // Disconnected unexpectedly from ACS, compute backoff duration to // reconnect reconnectDelay := acsSession . computeReconnectDelay ( isInactiveInstance ) seelog . Infof ( " " , reconnectDelay . String ( ) ) waitComplete := acsSession . waitForDuration ( reconnectDelay ) if waitComplete { // If the context was not cancelled and we've waited for the // wait duration without any errors, send the message to the channel // to reconnect to ACS sendEmptyMessageOnChannel ( connectToACS ) } else { // Wait was interrupted. We expect the session to close as canceling // the session context is the only way to end up here. Print a message // to indicate the same seelog . Info ( " " ) } } case <- acsSession . ctx . Done ( ) : seelog . Debugf ( " " ) return acsSession . ctx . Err ( ) } } }
// Start starts the session. It'll forever keep trying to connect to ACS unless
// the context is cancelled.
//
// If the context is cancelled, Start() would return with the error code returned
// by the context.
// If the instance is deregistered, Start() would emit an event to the
// deregister-instance event stream and sets the connection backoff time to 1 hour.
Start starts the session . It ll forever keep trying to connect to ACS unless the context is cancelled . If the context is cancelled Start () would return with the error code returned by the context . If the instance is deregistered Start () would emit an event to the deregister - instance event stream and sets the connection backoff time to 1 hour .