Method run

server/topic.go:117–281

(hub *Hub)

}

func (t *Topic) run(hub *Hub) {

	log.Printf("Topic started: '%s'", t.name)

	keepAlive := time.Second * 5 // TODO(gene): read keepalive value from the command line
	killTimer := time.NewTimer(time.Hour)
	killTimer.Stop()

	for {
		select {
		case sreg := <-t.reg:
			// Request to add a conection to this topic

			// The topic is alive, so stop the kill timer, if it's ticking. We don't want the topic to die
			// while processing the call
			killTimer.Stop()

			if err := t.handleSubscription(hub, sreg); err == nil {
				// give a broadcast channel to the connection (.read)
				// give channel to use when shutting down (.done)
				sreg.sess.subs[t.name] = &Subscription{broadcast: t.broadcast, done: t.unreg, meta: t.meta}
				t.sessions[sreg.sess] = true

			} else if len(t.sessions) == 0 {
				// Failed to subscribe, the topic is still inactive
				killTimer.Reset(keepAlive)
			}
		case leave := <-t.unreg:
			// Remove connection from topic; session may continue to function
			delete(t.sessions, leave.sess)

			now := time.Now().UTC().Round(time.Millisecond)
			pud := t.perUser[leave.sess.uid]
			if pud.lastSeenTag == nil {
				pud.lastSeenTag = map[string]time.Time{}
			}
			pud.lastSeenTag[leave.sess.tag] = now
			t.perUser[leave.sess.uid] = pud
			if err := store.Topics.UpdateLastSeen(t.appid, t.name, leave.sess.uid, leave.sess.tag, now); err != nil {
				log.Println(err)
			}

			// User wants to unsubscribe.
			if leave.unsub {
				// Delete user's subscription from the database
				if err := store.Subs.Delete(t.appid, t.name, leave.sess.uid); err != nil {
					if leave.pkt != nil {
						leave.sess.QueueOut(ErrUnknown(leave.pkt.Leave.Id, leave.pkt.Leave.Topic, now))
					}
					log.Println(err)
					continue
				}

				// Delete per-user data
				delete(t.perUser, leave.sess.uid)

				// Detach all user's sessions
				for sess, _ := range t.sessions {