func (w *sessionStreamHandler) stream(ws *websocket.Conn) error {
	w.ws = ws
	clt, err := w.site.GetClient()
	if err != nil {
		return trace.Wrap(err)
	}
	// spin up a goroutine to detect closed socket by reading
	// from it
	go func() {
		defer w.Close()
		io.Copy(ioutil.Discard, ws)
	}()

	eventsCursor := -1
	emptyEventList := make([]events.EventFields, 0)

	pollEvents := func() []events.EventFields {
		// ask for any events than happened since the last call:
		re, err := clt.GetSessionEvents(w.sessionID, eventsCursor+1)
		if err != nil {
			log.Error(err)
			return emptyEventList
		}
		batchLen := len(re)
		if batchLen == 0 {
			return emptyEventList
		}
		// advance the cursor, so next time we'll ask for the latest:
		eventsCursor = re[batchLen-1].GetInt(events.EventCursor)
		return re
	}

	ticker := time.NewTicker(w.pollPeriod)
	defer ticker.Stop()
	defer w.Close()

	// keep polling in a loop:
	for {
		// wait for next timer tick or a signal to abort:
		select {
		case <-ticker.C:
		case <-w.closeC:
			log.Infof("[web] session.stream() exited")
			return nil
		}

		newEvents := pollEvents()
		sess, err := clt.GetSession(w.sessionID)
		if err != nil {
			log.Error(err)
		}
		if sess == nil {
			log.Warningf("invalid session ID: %v", w.sessionID)
			continue
		}
		servers, err := clt.GetNodes()
		if err != nil {
			log.Error(err)
		}
		if len(newEvents) > 0 {
			log.Infof("[WEB] streaming for %v. Events: %v, Nodes: %v, Parties: %v",
				w.sessionID, len(newEvents), len(servers), len(sess.Parties))
		}

		// push events to the web client
		event := &sessionStreamEvent{
			Events:  newEvents,
			Session: sess,
			Servers: servers,
		}
		if err := websocket.JSON.Send(ws, event); err != nil {
			log.Error(err)
		}
	}
}
// stream runs in a loop generating "something changed" events for a
// given active WebSession
//
// The events are fed to a web client via the websocket
stream runs in a loop generating "something changed" events for a
given active WebSession

The events are fed to a web client via the websocket
// stream runs in a loop generating " something changed " events for a // given active WebSession // // The events are fed to a web client via the websocket
func (w *sessionStreamHandler) stream(ws *websocket.Conn) error {
	w.ws = ws
	clt, err := w.site.GetClient()
	if err != nil {
		return trace.Wrap(err)
	}
	go func() {
		defer w.Close()
		io.Copy(ioutil.Discard, ws)
	}()
	eventsCursor := -1
	emptyEventList := make([]events.EventFields, 0)
	pollEvents := func() []events.EventFields {
		re, err := clt.GetSessionEvents(w.sessionID, eventsCursor+1)
		if err != nil {
			log.Error(err)
			return emptyEventList
		}
		batchLen := len(re)
		if batchLen == 0 {
			return emptyEventList
		}
		eventsCursor = re[batchLen-1].GetInt(events.EventCursor)
		return re
	}
	ticker := time.NewTicker(w.pollPeriod)
	defer ticker.Stop()
	defer w.Close()
	for {
		select {
		case <-ticker.C:
		case <-w.closeC:
			log.Infof("[web] session.stream() exited")
			return nil
		}
		newEvents := pollEvents()
		sess, err := clt.GetSession(w.sessionID)
		if err != nil {
			log.Error(err)
		}
		if sess == nil {
			log.Warningf("invalid session ID: %v", w.sessionID)
			continue
		}
		servers, err := clt.GetNodes()
		if err != nil {
			log.Error(err)
		}
		if len(newEvents) > 0 {
			log.Infof("[WEB] streaming for %v. Events: %v, Nodes: %v, Parties: %v",
				w.sessionID, len(newEvents), len(servers), len(sess.Parties))
		}
		event := &sessionStreamEvent{
			Events:  newEvents,
			Session: sess,
			Servers: servers,
		}
		if err := websocket.JSON.Send(ws, event); err != nil {
			log.Error(err)
		}
	}
}
func ( w * sessionStreamHandler ) stream ( ws * websocket . Conn ) error { w . ws = ws clt , err := w . site . GetClient ( ) if err != nil { return trace . Wrap ( err ) } go func ( ) { defer w . Close ( ) io . Copy ( ioutil . Discard , ws ) } ( ) eventsCursor := - 1 emptyEventList := make ( [ ] events . EventFields , 0 ) pollEvents := func ( ) [ ] events . EventFields { re , err := clt . GetSessionEvents ( w . sessionID , eventsCursor + 1 ) if err != nil { log . Error ( err ) return emptyEventList } batchLen := len ( re ) if batchLen == 0 { return emptyEventList } eventsCursor = re [ batchLen - 1 ] . GetInt ( events . EventCursor ) return re } ticker := time . NewTicker ( w . pollPeriod ) defer ticker . Stop ( ) defer w . Close ( ) for { select { case <- ticker . C : case <- w . closeC : log . Infof ( " " ) return nil } newEvents := pollEvents ( ) sess , err := clt . GetSession ( w . sessionID ) if err != nil { log . Error ( err ) } if sess == nil { log . Warningf ( " " , w . sessionID ) continue } servers , err := clt . GetNodes ( ) if err != nil { log . Error ( err ) } if len ( newEvents ) > 0 { log . Infof ( " " , w . sessionID , len ( newEvents ) , len ( servers ) , len ( sess . Parties ) ) } event := & sessionStreamEvent { Events : newEvents , Session : sess , Servers : servers , } if err := websocket . JSON . Send ( ws , event ) ; err != nil { log . Error ( err ) } } }
stream runs in a loop generating "something changed" events for a
given active WebSession
stream runs in a loop generating " something changed " events for a given active WebSession
// spin up a goroutine to detect closed socket by reading // from it // ask for any events than happened since the last call: // advance the cursor, so next time we'll ask for the latest: // keep polling in a loop: // wait for next timer tick or a signal to abort: // push events to the web client