Simple Go-based HTTP streaming via HTTP and websockets.

client.go

package main

import (
	"encoding/json"
	"flag"
	"io"
	"net/http"

	"github.com/golang/glog"
	"golang.org/x/net/websocket"
)

var useWebsockets = flag.Bool("websockets", false, "Whether to use websockets")

type Message struct {
	Id      int    `json:"id,omitempty"`
	Message string `json:"message,omitempty"`
}

// Client.
func main() {
	flag.Parse()

	if *useWebsockets {
		ws, err := websocket.Dial("ws://localhost:8080/", "", "http://localhost:8080")
		for {
			var m Message
			err = websocket.JSON.Receive(ws, &m)
			if err != nil {
				if err == io.EOF {
					break
				}
				glog.Fatal(err)
			}
			glog.Infof("Received: %+v", m)
		}
	} else {
		glog.Info("Sending request...")
		req, err := http.NewRequest("GET", "http://localhost:8080", nil)
		if err != nil {
			glog.Fatal(err)
		}
		resp, err := http.DefaultClient.Do(req)
		if err != nil {
			glog.Fatal(err)
		}
		if resp.StatusCode != http.StatusOK {
			glog.Fatalf("Status code is not OK: %v (%s)", resp.StatusCode, resp.Status)
		}

		dec := json.NewDecoder(resp.Body)
		for {
			var m Message
			err := dec.Decode(&m)
			if err != nil {
				if err == io.EOF {
					break
				}
				glog.Fatal(err)
			}
			glog.Infof("Got response: %+v", m)
		}
	}

	glog.Infof("Server finished request...")
}

server.go

package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"net/http"
	"regexp"
	"strings"
	"time"

	"github.com/golang/glog"
	"golang.org/x/net/websocket"
)

type Message struct {
	Id      int    `json:"id,omitempty"`
	Message string `json:"message,omitempty"`
}

// Heavily based on Kubernetes' (https://github.com/GoogleCloudPlatform/kubernetes) detection code.
var connectionUpgradeRegex = regexp.MustCompile("(^|.*,\\s*)upgrade($|\\s*,)")

func isWebsocketRequest(req *http.Request) bool {
	return connectionUpgradeRegex.MatchString(strings.ToLower(req.Header.Get("Connection"))) && strings.ToLower(req.Header.Get("Upgrade")) == "websocket"
}

func Handle(w http.ResponseWriter, r *http.Request) {
	// Handle websockets if specified.
	if isWebsocketRequest(r) {
		websocket.Handler(HandleWebSockets).ServeHTTP(w, r)
	} else {
		HandleHttp(w, r)
	}
	glog.Info("Finished sending response...")
}

func HandleWebSockets(ws *websocket.Conn) {
	for i := 0; i < 5; i++ {
		glog.Infof("Sending some data: %d", i)
		m := Message{
			Id:      i,
			Message: fmt.Sprintf("Sending you \"%d\"", i),
		}
		err := websocket.JSON.Send(ws, &m)
		if err != nil {
			glog.Infof("Client stopped listening...")
			return
		}

		// Artificially induce a 1s pause
		time.Sleep(time.Second)
	}
}

func HandleHttp(w http.ResponseWriter, r *http.Request) {
	cn, ok := w.(http.CloseNotifier)
	if !ok {
		http.NotFound(w, r)
		return
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.NotFound(w, r)
		return
	}

	// Send the initial headers saying we're gonna stream the response.
	w.Header().Set("Transfer-Encoding", "chunked")
	w.WriteHeader(http.StatusOK)
	flusher.Flush()

	enc := json.NewEncoder(w)

	for i := 0; i < 5; i++ {
		select {
		case <-cn.CloseNotify():
			glog.Infof("Client stopped listening")
			return
		default:
			// Artificially wait a second between reponses.
			time.Sleep(time.Second)

			glog.Infof("Sending some data: %d", i)
			m := Message{
				Id:      i,
				Message: fmt.Sprintf("Sending you \"%d\"", i),
			}

			// Send some data.
			err := enc.Encode(m)
			if err != nil {
				glog.Fatal(err)
			}
			flusher.Flush()
		}
	}
}

// Server.
func main() {
	flag.Parse()

	http.HandleFunc("/", Handle)

	glog.Infof("Serving...")
	glog.Fatal(http.ListenAndServe(":8080", nil))
}