Merge pull request #156 from zorglube/viewers_stats

Stream Viewers stats

chatclient.go

@@ -91,17 +91,20 @@ func (cl *Client) NewMsg(data common.ClientData) {
 		// because the /me command outputs to the messages
 		msg = addSpoilerTags(msg)
 
+		msgLen := len(msg)
+
 		// Don't send zero-length messages
-		if len(msg) == 0 {
+		if msgLen == 0 {
 			return
 		}
 
 		if strings.HasPrefix(msg, "/") {
 			// is a command
-			msg = msg[1:len(msg)]
+			msg = msg[1:msgLen]
 			fullcmd := strings.Split(msg, " ")
 			cmd := strings.ToLower(fullcmd[0])
-			args := fullcmd[1:len(fullcmd)]
+			fullcmdLen := len(fullcmd)
+			args := fullcmd[1:fullcmdLen]
 
 			response, err := commands.RunCommand(cmd, args, cl)
 			if response != "" || err != nil {
@@ -136,7 +139,7 @@ func (cl *Client) NewMsg(data common.ClientData) {
 			}
 
 			// Trim long messages
-			if len(msg) > 400 {
+			if msgLen > 400 {
 				msg = msg[0:400]
 			}
 

chatcommands.go

@@ -277,11 +277,13 @@ var commands = &CommandControl{
 				cl.belongsTo.clientsMtx.Unlock()
 
 				// Just print max users and time alive here
-				return fmt.Sprintf("Current users in chat: <b>%d</b><br />Max users in chat: <b>%d</b><br />Server uptime: <b>%s</b><br />Stream uptime: <b>%s</b>",
+				return fmt.Sprintf("Current users in chat: <b>%d</b><br />Max users in chat: <b>%d</b><br />Server uptime: <b>%s</b><br />Stream uptime: <b>%s</b><br />Viewers: <b>%d</b><br />Max Viewers: <b>%d</b>",
 					users,
 					stats.getMaxUsers(),
 					time.Since(stats.start),
 					stats.getStreamLength(),
+					stats.getViewerCount(),
+					stats.getMaxViewerCount(),
 				), nil
 			},
 		},

common/utils.go

@@ -3,6 +3,7 @@ package common
 // Misc utils
 
 import (
+	"net/http"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -41,3 +42,20 @@ func Substr(input string, start int, length int) string {
 
 	return string(asRunes[start : start+length])
 }
+
+// Return the value of "Forwarded" or "X-Forwarded-For",
+// if "Forwarded" & "X-Forwarded-For" are present then "Forwarded" value is returned.
+// Return "" if "Forwarded" and "X-Forwarded-For" are absent.
+func ExtractForwarded(r *http.Request) string {
+	f := r.Header.Get("Forwarded")
+	if f != "" {
+		return f
+	}
+
+	xff := r.Header.Get("X-Forwarded-For")
+	if xff != "" {
+		return xff
+	}
+
+	return ""
+}

handlers.go

@@ -104,7 +104,7 @@ func wsHandler(w http.ResponseWriter, r *http.Request) {
 		Conn: conn,
 		// If the server is behind a reverse proxy (eg, Nginx), look
 		// for this header to get the real IP address of the client.
-		forwardedFor: r.Header.Get("X-Forwarded-For"),
+		forwardedFor: common.ExtractForwarded(r),
 	}
 
 	go func() {
@@ -424,9 +424,14 @@ func handleLive(w http.ResponseWriter, r *http.Request) {
 		muxer := flv.NewMuxerWriteFlusher(writeFlusher{httpflusher: flusher, Writer: w})
 		cursor := ch.que.Latest()
 
+		session, _ := sstore.Get(r, "moviesession")
+		stats.addViewer(session.ID)
 		avutil.CopyFile(muxer, cursor)
+		stats.removeViewer(session.ID)
 	} else {
-
+		// Maybe HTTP_204 is better than HTTP_404
+		w.WriteHeader(http.StatusNoContent)
+		stats.resetViewers()
 	}
 }
 

stats.go

@@ -8,18 +8,50 @@ import (
 )
 
 type streamStats struct {
-	messageIn  int
-	messageOut int
-	maxUsers   int
-	start      time.Time
-	mutex      sync.Mutex
-
+	messageIn   int
+	messageOut  int
+	maxUsers    int
+	start       time.Time
+	mutex       sync.Mutex
 	streamStart time.Time
 	streamLive  bool // True if live
+	viewers     map[string]int
+	maxViewers  int
+}
+
+func (s *streamStats) addViewer(id string) {
+	s.mutex.Lock()
+	s.viewers[id] = len(s.viewers)
+	size := len(s.viewers)
+	s.updateMaxViewers(size)
+	s.mutex.Unlock()
+
+	common.LogDebugf("[stats] %d viewer(s) connected\n", size)
+}
+func (s *streamStats) removeViewer(id string) {
+	s.mutex.Lock()
+	delete(s.viewers, id)
+	s.mutex.Unlock()
+
+	common.LogDebugf("[stats] One viewer left the stream\n")
+}
+
+func (s *streamStats) updateMaxViewers(size int) {
+	if s.maxViewers < size {
+		s.maxViewers = size
+	}
+}
+
+func (s *streamStats) resetViewers() {
+	s.viewers = sessionsMapNew()
+}
+
+func sessionsMapNew() map[string]int {
+	return make(map[string]int)
 }
 
 func newStreamStats() streamStats {
-	return streamStats{start: time.Now(), streamLive: false}
+	return streamStats{start: time.Now(), streamLive: false, viewers: sessionsMapNew()}
 }
 
 func (s *streamStats) msgInInc() {
@@ -53,10 +85,11 @@ func (s *streamStats) Print() {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
-	common.LogInfof("Messages In: %d\n", s.messageIn)
-	common.LogInfof("Messages Out: %d\n", s.messageOut)
-	common.LogInfof("Max users in chat: %d\n", s.maxUsers)
-	common.LogInfof("Total Time: %s\n", time.Since(s.start))
+	common.LogInfof("[stats] Messages In: %d\n", s.messageIn)
+	common.LogInfof("[stats] Messages Out: %d\n", s.messageOut)
+	common.LogInfof("[stats] Max users in chat: %d\n", s.maxUsers)
+	common.LogInfof("[stats] Total Time: %s\n", time.Since(s.start))
+	common.LogInfof("[stats] Max Stream Viewer: %d\n", s.maxViewers)
 }
 
 func (s *streamStats) startStream() {
@@ -83,3 +116,24 @@ func (s *streamStats) getStreamLength() time.Duration {
 	}
 	return time.Since(s.streamStart)
 }
+
+func (s *streamStats) getViewerCount() int {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	return len(s.viewers)
+}
+
+func (s *streamStats) getMaxViewerCount() int {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	return s.maxViewers
+}
+
+func (s *streamStats) getViewers() map[string]int {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	return s.viewers
+}