notify/client_session_route.go

package notify

import "context"

// clientSessionRoute pins a send or inbound dispatch to one physical client
// transport. The logical client session can survive a transport reattach, so
// the epoch alone is not sufficient to prevent an old operation from crossing
// onto the replacement connection.
type clientSessionRoute struct {
	runtime          *clientSessionRuntime
	binding          *transportBinding
	epoch            uint64
	sessionStopCtx   context.Context
	transportStopCtx context.Context
}

func clientSessionRouteFromRuntime(rt *clientSessionRuntime) clientSessionRoute {
	if rt == nil {
		return clientSessionRoute{}
	}
	transportStopCtx := rt.transportStopCtx
	if transportStopCtx == nil {
		transportStopCtx = rt.stopCtx
	}
	return clientSessionRoute{
		runtime:          rt,
		binding:          rt.transport,
		epoch:            rt.epoch,
		sessionStopCtx:   rt.stopCtx,
		transportStopCtx: transportStopCtx,
	}
}

func clientSessionRouteContext(route clientSessionRoute) context.Context {
	if route.transportStopCtx != nil {
		return route.transportStopCtx
	}
	return route.sessionStopCtx
}

func sameClientSessionRoute(left, right clientSessionRoute) bool {
	if left.runtime != nil || right.runtime != nil {
		return left.runtime != nil && left.runtime == right.runtime
	}
	if left.binding != nil || right.binding != nil {
		return left.binding != nil && left.binding == right.binding &&
			(left.epoch == 0 || right.epoch == 0 || left.epoch == right.epoch)
	}
	return left.epoch != 0 && left.epoch == right.epoch
}

func (r clientSessionRoute) bound() bool {
	return r.binding != nil || r.epoch != 0
}

func (r clientSessionRoute) inboundQueueKey() interface{} {
	if r.binding != nil {
		return r.binding
	}
	return "b612"
}

func (c *ClientCommon) clientSessionRouteSnapshot() clientSessionRoute {
	if c == nil {
		return clientSessionRoute{}
	}
	return clientSessionRouteFromRuntime(c.clientSessionRuntimeSnapshot())
}

func (c *ClientCommon) clientSessionRouteCurrent(route clientSessionRoute) bool {
	if c == nil {
		return false
	}
	if !route.bound() {
		return true
	}
	current := c.clientSessionRuntimeSnapshot()
	if current == nil {
		return false
	}
	if route.epoch != 0 && current.epoch != route.epoch {
		return false
	}
	return route.binding != nil && current.transport == route.binding
}

func (c *ClientCommon) ensureClientSessionRouteSendReady(route clientSessionRoute) error {
	if err := c.ensureClientSendReady(); err != nil {
		return err
	}
	if !c.clientSessionRouteCurrent(route) {
		return transportDetachedSessionEpochError()
	}
	if route.transportStopCtx != nil {
		select {
		case <-route.transportStopCtx.Done():
			if route.sessionStopCtx != nil && route.sessionStopCtx.Err() != nil {
				return errServiceShutdown
			}
			return transportDetachedSessionEpochError()
		default:
		}
	}
	if route.binding == nil {
		return clientTransportDetachedError(c)
	}
	return nil
}