Method runProxy

server/topic_proxy.go:17–117

(hub *Hub)

)

func (t *Topic) runProxy(hub *Hub) {
	killTimer := time.NewTimer(time.Hour)
	killTimer.Stop()

	for {
		select {
		case msg := <-t.reg:
			// Request to add a connection to this topic
			if t.isInactive() {
				msg.sess.queueOut(ErrLockedReply(msg, types.TimeNow()))
			} else if err := globals.cluster.routeToTopicMaster(ProxyReqJoin, msg, t.name, msg.sess); err != nil {
				// Response (ctrl message) will be handled when it's received via the proxy channel.
				logs.Warn.Printf("proxy topic[%s]: route join request from proxy to master failed - %s", t.name, err)
				msg.sess.queueOut(ErrClusterUnreachableReply(msg, types.TimeNow()))
			}
			if msg.sess.inflightReqs != nil {
				msg.sess.inflightReqs.Done()
			}

		case msg := <-t.unreg:
			if !t.handleProxyLeaveRequest(msg, killTimer) {
				sid := "nil"
				if msg.sess != nil {
					sid = msg.sess.sid
				}
				logs.Warn.Printf("proxy topic[%s]: failed to update proxy topic state for leave request - sid %s", t.name, sid)
				msg.sess.queueOut(ErrClusterUnreachableReply(msg, types.TimeNow()))
			}
			if msg.init && msg.sess.inflightReqs != nil {
				// If it's a client initiated request.
				msg.sess.inflightReqs.Done()
			}

		case msg := <-t.clientMsg:
			// Content message intended for broadcasting to recipients
			if err := globals.cluster.routeToTopicMaster(ProxyReqBroadcast, msg, t.name, msg.sess); err != nil {
				logs.Warn.Printf("topic proxy[%s]: route broadcast request from proxy to master failed - %s", t.name, err)
				msg.sess.queueOut(ErrClusterUnreachableReply(msg, types.TimeNow()))
			}

		case msg := <-t.serverMsg:
			if msg.Info != nil || msg.Pres != nil {
				globals.cluster.routeToTopicIntraCluster(t.name, msg, msg.sess)
			} else {
				// FIXME: should something be done here?
				logs.Err.Printf("ERROR!!! topic proxy[%s]: unexpected server-side message in proxy topic %s", t.name, msg.describe())
			}

		case msg := <-t.meta:
			// Request to get/set topic metadata
			if err := globals.cluster.routeToTopicMaster(ProxyReqMeta, msg, t.name, msg.sess); err != nil {
				logs.Warn.Printf("proxy topic[%s]: route meta request from proxy to master failed - %s", t.name, err)
				msg.sess.queueOut(ErrClusterUnreachableReply(msg, types.TimeNow()))
			}

		case upd := <-t.supd:
			// Either an update to 'me' user agent from one of the sessions or
			// background session comes to foreground.