examples/sse.go
func SendEvents(w http.ResponseWriter, r *http.Request, channels []string) {
	// make sure that the writer supports flushing
	flusher, ok := w.(http.Flusher)

	if !ok {
		log.Error("current response %T does not implement http.Flusher, plase check your middlewares that wraps response", w)
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}

	closeNotifier, ok := w.(http.CloseNotifier)
	if !ok {
		log.Error("current response %T does not implement http.CloseNotifier, plase check your middlewares that wraps response", w)
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}

	sub, err := pubsub.Subscribe(channels)
	if err != nil {
		log.Error("subscribe failed: %v", err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	// set the headers related to event streaming
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")

	// channel to send data to client
	var send = make(chan []byte)
	var closeConn = closeNotifier.CloseNotify()
	var heartbeatTicker = time.NewTicker(10 * time.Second) // TODO get from config
	var heatbeat = []byte{}

	var stop = func() {
		log.Info("SSE streaming stopped")
		heartbeatTicker.Stop()
		sub.Close()
		close(send)
	}

	// goroutine to listen all room events
	go func() {
		defer func() {
			if err := recover(); err != nil {
				log.Error("recovered from panic: %+v", err)
				debug.PrintStack()
			}
		}()

		log.Info("SSE streaming started")

		for {
			select {
			case msg := <-sub.Read():
				var err error
				var data, ok = msg.([]byte)
				if !ok {
					data, err = json.Marshal(msg)
					if err != nil {
						// TODO should we ignore error messages?
						log.Error("json.Marshal failed with: %+v", err)
						continue
					}
				}
				if len(data) == 0 {
					log.Warning("empty message is not allowed")
					continue
				}
				send <- data
			// listen to connection close and un-register message channel
			case <-sub.CloseNotify():
				log.Info("subscription closed")
				stop()
				return
			case <-closeConn:
				stop()
				return
			case <-heartbeatTicker.C:
				send <- heatbeat
			}
		}
	}()

	for {
		var data, ok = <-send
		if !ok {
			log.Info("connection closed, stop streaming of %v", channels)
			return
		}

		if len(data) == 0 {
			fmt.Fprint(w, ":heartbeat signal\n\n")
		} else {
			fmt.Fprintf(w, "data: %s\n\n", data)
		}

		flusher.Flush()
	}
}
func SendEvents(w http.ResponseWriter, r *http.Request, channels []string) {
	// make sure that the writer supports flushing
	flusher, ok := w.(http.Flusher)

	if !ok {
		log.Error("current response %T does not implement http.Flusher, plase check your middlewares that wraps response", w)
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}

	closeNotifier, ok := w.(http.CloseNotifier)
	if !ok {
		log.Error("current response %T does not implement http.CloseNotifier, plase check your middlewares that wraps response", w)
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}

	sub, err := pubsub.Subscribe(channels)
	if err != nil {
		log.Error("subscribe failed: %v", err)
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	// set the headers related to event streaming
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("Connection", "keep-alive")

	// channel to send data to client
	var send = make(chan []byte)
	var closeConn = closeNotifier.CloseNotify()
	var heartbeatTicker = time.NewTicker(10 * time.Second) // TODO get from config
	var heatbeat = []byte{}

	var stop = func() {
		log.Info("SSE streaming stopped")
		heartbeatTicker.Stop()
		sub.Close()
		close(send)
	}

	// goroutine to listen all room events
	go func() {
		defer func() {
			if err := recover(); err != nil {
				log.Error("recovered from panic: %+v", err)
				debug.PrintStack()
			}
		}()

		log.Info("SSE streaming started")

		for {
			select {
			case msg := <-sub.Read():
				var err error
				var data, ok = msg.([]byte)
				if !ok {
					data, err = json.Marshal(msg)
					if err != nil {
						// TODO should we ignore error messages?
						log.Error("json.Marshal failed with: %+v", err)
						continue
					}
				}
				if len(data) == 0 {
					log.Warning("empty message is not allowed")
					continue
				}
				send <- data
			// listen to connection close and un-register message channel
			case <-sub.CloseNotify():
				log.Info("subscription closed")
				stop()
				return
			case <-closeConn:
				stop()
				return
			case <-heartbeatTicker.C:
				send <- heatbeat
			}
		}
	}()

	for {
		var data, ok = <-send
		if !ok {
			log.Info("connection closed, stop streaming of %v", channels)
			return
		}

		if len(data) == 0 {
			fmt.Fprint(w, ":heartbeat signal\n\n")
		} else {
			fmt.Fprintf(w, "data: %s\n\n", data)
		}

		flusher.Flush()
	}
}
func SendEvents ( w http . ResponseWriter , r * http . Request , channels [ ] string ) { // make sure that the writer supports flushing flusher , ok := w . ( http . Flusher ) if ! ok { log . Error ( " " , w ) http . Error ( w , " " , http . StatusInternalServerError ) return } closeNotifier , ok := w . ( http . CloseNotifier ) if ! ok { log . Error ( " " , w ) http . Error ( w , " " , http . StatusInternalServerError ) return } sub , err := pubsub . Subscribe ( channels ) if err != nil { log . Error ( " " , err ) http . Error ( w , err . Error ( ) , http . StatusInternalServerError ) return } // set the headers related to event streaming w . Header ( ) . Set ( " " , " " ) w . Header ( ) . Set ( " " , " " ) w . Header ( ) . Set ( " " , " " ) // channel to send data to client var send = make ( chan [ ] byte ) var closeConn = closeNotifier . CloseNotify ( ) var heartbeatTicker = time . NewTicker ( 10 * time . Second ) // TODO get from config var heatbeat = [ ] byte { } var stop = func ( ) { log . Info ( " " ) heartbeatTicker . Stop ( ) sub . Close ( ) close ( send ) } // goroutine to listen all room events go func ( ) { defer func ( ) { if err := recover ( ) ; err != nil { log . Error ( " " , err ) debug . PrintStack ( ) } } ( ) log . Info ( " " ) for { select { case msg := <- sub . Read ( ) : var err error var data , ok = msg . ( [ ] byte ) if ! ok { data , err = json . Marshal ( msg ) if err != nil { // TODO should we ignore error messages? log . Error ( " " , err ) continue } } if len ( data ) == 0 { log . Warning ( " " ) continue } send <- data // listen to connection close and un-register message channel case <- sub . CloseNotify ( ) : log . Info ( " " ) stop ( ) return case <- closeConn : stop ( ) return case <- heartbeatTicker . C : send <- heatbeat } } } ( ) for { var data , ok = <- send if ! ok { log . Info ( " " , channels ) return } if len ( data ) == 0 { fmt . Fprint ( w , " \n \n " ) } else { fmt . Fprintf ( w , " \n \n " , data ) } flusher . Flush ( ) } }
// SendEvents streams events from specified channels as Server Sent Events packets
SendEvents streams events from specified channels as Server Sent Events packets