First step towards handler modularization. Extracted CLI and Websocket handlers.

Author: sergi

client.go

@@ -42,7 +42,7 @@ func getHostName() string {
 	return hostname
 }
 
-// Init initializes the client to start sending Chunks at the given address
+// Listen initializes the client to start sending Chunks at the given address
 func Init(opts ConsumerOptions, stream io.Reader, conn io.Writer) error {
 	if opts.Id == "" {
 		opts.Id, _ = shortid.Generate()

cmd/siph/main.go

@@ -8,6 +8,8 @@ import (
 	"os"
 
 	"github.com/sergi/siphon"
+	"github.com/sergi/siphon/handlers/cli"
+	"github.com/sergi/siphon/handlers/websocket"
 	"gopkg.in/alecthomas/kingpin.v2"
 )
 
@@ -32,7 +34,7 @@ var (
 	app = kingpin.New("siph", "A real-time stream utility.")
 
 	server        = app.Command("server", "Run in Server mode")
-	serverUDPPort = server.Flag("udp-port", "UDP port to run in Server mode").Short('u').Int()
+	serverUDPPort = server.Flag("udp-port", "UDP port to run in Server mode").Short('u').Default("1200").Int()
 	serverWsPort  = server.Flag("ws-port", "WebSockets port to run in Server mode").Short('w').Int()
 
 	client         = app.Command("client", "Stream output to a server")
@@ -70,9 +72,8 @@ func main() {
 	switch kingpin.MustParse(app.Parse(os.Args[1:])) {
 
 	case server.FullCommand():
-		server := siphon.NewServer(*serverUDPPort, *serverWsPort)
 		fmt.Print(siphonTemplate)
-		err := server.Init()
+		_, err := siphon.NewServer(siphon.Port(*serverUDPPort), cli.New(), websocket.New())
 		if err != nil {
 			fmt.Println(err)
 			os.Exit(ExitCodeError)

handlers/cli/cli.go

@@ -0,0 +1,56 @@
+package cli
+
+import (
+	"fmt"
+	"io"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/sergi/siphon"
+)
+
+const pingPeriod = 5 * time.Minute
+
+type Handler struct {
+	mu     sync.Mutex
+	Writer io.Writer
+}
+
+func New(options ...func(*Handler)) func(*siphon.Server) {
+	return func(server *siphon.Server) {
+		h := &Handler{
+			Writer: os.Stdout,
+		}
+
+		for _, option := range options {
+			option(h)
+		}
+
+		go h.handleConnection(server)
+	}
+}
+
+func (h *Handler) handleConnection(s *siphon.Server) {
+	sub := s.Subscribe(siphon.DefaultChannel)
+	t := time.NewTicker(pingPeriod)
+
+	var message []byte
+
+	for {
+		select {
+		case <-t.C:
+			message = nil
+		case message = <-sub:
+		}
+
+		_, err := h.Writer.Write(message)
+		if err != nil {
+			fmt.Print(err)
+			break
+		}
+	}
+
+	t.Stop()
+	s.Unsubscribe(siphon.DefaultChannel, sub)
+}

handlers/cli/cli_test.go

@@ -0,0 +1,40 @@
+package cli
+
+import (
+	"fmt"
+	"path/filepath"
+	"reflect"
+	"runtime"
+	"testing"
+)
+
+// assert fails the test if the condition is false.
+func assert(tb testing.TB, condition bool, msg string, v ...interface{}) {
+	if !condition {
+		_, file, line, _ := runtime.Caller(1)
+		fmt.Printf("\033[31m%s:%d: "+msg+"\033[39m\n\n", append([]interface{}{filepath.Base(file), line}, v...)...)
+		tb.FailNow()
+	}
+}
+
+// ok fails the test if an err is not nil.
+func ok(tb testing.TB, err error) {
+	if err != nil {
+		_, file, line, _ := runtime.Caller(1)
+		fmt.Printf("\033[31m%s:%d: unexpected error: %s\033[39m\n\n", filepath.Base(file), line, err.Error())
+		tb.FailNow()
+	}
+}
+
+// equals fails the test if exp is not equal to act.
+func equals(tb testing.TB, exp, act interface{}) {
+	if !reflect.DeepEqual(exp, act) {
+		_, file, line, _ := runtime.Caller(1)
+		fmt.Printf("\033[31m%s:%d:\n\n\texp: %#v\n\n\tgot: %#v\033[39m\n\n", filepath.Base(file), line, exp, act)
+		tb.FailNow()
+	}
+}
+
+func TestNewCli(t *testing.T) {
+
+}

handlers/websocket/websocket.go

@@ -0,0 +1,99 @@
+package websocket
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+	"strconv"
+	"sync/atomic"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/sergi/siphon"
+)
+
+const defaultWsPort = 3000
+
+const (
+	maxMessageSize = 1024
+	pingPeriod     = 5 * time.Minute
+)
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: maxMessageSize,
+	CheckOrigin: func(r *http.Request) bool { // WARNING! That's probably messed up https://stackoverflow.com/questions/33323337/update-send-the-origin-header-with-js-websockets
+		return true
+	},
+}
+
+type Handler struct {
+	connected int64
+	failed    int64
+	wsPort    int
+}
+
+func New(options ...func(*Handler)) func(*siphon.Server) {
+	return func(server *siphon.Server) {
+		h := &Handler{
+			wsPort: defaultWsPort,
+		}
+
+		for _, option := range options {
+			option(h)
+		}
+
+		http.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) {
+			ws, err := upgrader.Upgrade(w, r, nil)
+			if err != nil {
+				fmt.Println(err)
+				return
+			}
+			// channel := r.URL.Path[1:]
+			// launch a new goroutine so that this function can return and the http server can free up
+			// buffers associated with this connection
+			go h.handleConnection(ws, server)
+		})
+
+		go func() {
+			log.Println("Initializing Websocket Handler at localhost:" + strconv.Itoa(h.wsPort) + "/ws")
+			if err := http.ListenAndServe(":"+strconv.Itoa(h.wsPort), nil); err != nil {
+				log.Fatal(err)
+			}
+		}()
+	}
+}
+
+func Port(port int) func(*Handler) error {
+	return func(h *Handler) error {
+		h.wsPort = port
+		return nil
+	}
+}
+
+func (h *Handler) handleConnection(ws *websocket.Conn, s *siphon.Server) {
+	atomic.AddInt64(&h.connected, 1)
+	sub := s.Subscribe(siphon.DefaultChannel)
+	t := time.NewTicker(pingPeriod)
+
+	var message []byte
+
+	for {
+		select {
+		case <-t.C:
+			message = nil
+		case message = <-sub:
+		}
+
+		ws.SetWriteDeadline(time.Now().Add(30 * time.Second))
+		if err := ws.WriteMessage(websocket.TextMessage, message); err != nil {
+			break
+		}
+	}
+	atomic.AddInt64(&h.connected, -1)
+	atomic.AddInt64(&h.failed, 1)
+
+	t.Stop()
+	ws.Close()
+	s.Unsubscribe(siphon.DefaultChannel, sub)
+}