Behavision/desktop/stream_remote.go

package main

// Watching a camera in another building, from the app.
//
// The shop PC sits behind a router with no inbound route, so nothing here can
// pull its MJPEG stream - that stream is served on the shop PC's own loopback
// and always will be. Head office's LiveHub is the way round it: the agent
// asks outbound whether anyone is watching and pushes JPEG frames up for
// exactly as long as somebody is. The head-office web app already consumes
// that; this is the same feed, for the app.
//
// It arrives as base64 frames over SSE, which an <img> cannot render, so this
// re-emits them as multipart MJPEG - which an <img> renders natively, through
// the relay that already exists for the local engine. That is what keeps ONE
// code path in the screens: a tile points at a loopback URL and does not know
// or care which building the picture came from.

import (
	"bufio"
	"context"
	"encoding/base64"
	"fmt"
	"net/http"
	"strings"
	"time"
)

// The boundary is ours to choose; it only has to be a string the JPEG bytes
// cannot contain, and a marker line never appears inside JPEG data.
const mjpegBoundary = "behavisionframe"

// A frame is base64, so ~1.33 bytes on the wire per byte of picture. The
// engine re-encodes to 640 px for the relay and those measure ~20 KB, so this
// is roughly a hundredfold headroom - large enough never to clip a real frame
// and small enough that a broken or hostile stream cannot grow this process's
// memory without bound.
const maxFrameLine = 8 << 20

func (p *streamProxy) relayRemote(w http.ResponseWriter, r *http.Request,
	cameraID string, open func(context.Context, string) (*http.Response, error)) {

	w.Header().Set("Content-Type", "multipart/x-mixed-replace; boundary="+mjpegBoundary)
	w.Header().Set("Cache-Control", "no-store")
	flusher, _ := w.(http.Flusher)

	// Send the headers NOW, before any frame exists. Go writes them on the
	// first body write, so without this the whole response - status line
	// included - waits for the shop computer to start pushing, and a viewer
	// whose camera is slow to answer sees the REQUEST time out rather than a
	// stream that has not painted yet. Measured against production: 30
	// seconds and not even a Content-Type.
	if flusher != nil {
		flusher.Flush()
	}

	// Reconnecting is normal, not an error. The server caps one push at five
	// minutes so that a tab left open for a week cannot leave a shop
	// uploading for a week - so a viewer who IS still there simply asks
	// again. Doing it here rather than in the page is what lets the <img>
	// survive the cap: it never sees the stream end.
	sent := 0
	for {
		if r.Context().Err() != nil {
			return
		}
		n, err := p.pumpRemote(w, flusher, r.Context(), cameraID, open)
		sent += n
		if r.Context().Err() != nil {
			return
		}
		// Nothing was written and the attempt failed. Writing an error body
		// now would be writing it into a multipart stream the <img> is
		// already parsing, so the picture simply stays on whatever it last
		// showed and the screen's own "not connecting" state is the report.
		if err != nil && sent == 0 {
			return
		}
		select {
		case <-r.Context().Done():
			return
		case <-time.After(1500 * time.Millisecond):
		}
	}
}

// pumpRemote runs one SSE connection to exhaustion and returns how many
// frames it forwarded.
func (p *streamProxy) pumpRemote(w http.ResponseWriter, flusher http.Flusher,
	ctx context.Context, cameraID string,
	open func(context.Context, string) (*http.Response, error)) (int, error) {

	resp, err := open(ctx, cameraID)
	if err != nil {
		return 0, err
	}
	defer resp.Body.Close()

	sc := bufio.NewScanner(resp.Body)
	sc.Buffer(make([]byte, 0, 64*1024), maxFrameLine)

	var event, data string
	frames := 0
	for sc.Scan() {
		line := sc.Text()
		switch {
		case strings.HasPrefix(line, "event: "):
			event = strings.TrimSpace(line[7:])
		case strings.HasPrefix(line, "data: "):
			data = line[6:]
		case line == "":
			// End of one SSE event. `waiting` means head office has us
			// registered and the shop PC has not started pushing yet - a real
			// second or two while the agent is asked, and nothing to draw.
			if event == "frame" && data != "" {
				if err := writeMJPEGFrame(w, flusher, data); err != nil {
					return frames, err // the webview went away
				}
				frames++
			}
			event, data = "", ""
		}
	}
	return frames, sc.Err()
}

func writeMJPEGFrame(w http.ResponseWriter, flusher http.Flusher, b64 string) error {
	jpg, err := base64.StdEncoding.DecodeString(b64)
	if err != nil || len(jpg) == 0 {
		// One malformed frame is not a reason to tear down a working view.
		return nil
	}
	if _, err := fmt.Fprintf(w,
		"--%s\r\nContent-Type: image/jpeg\r\nContent-Length: %d\r\n\r\n",
		mjpegBoundary, len(jpg)); err != nil {
		return err
	}
	if _, err := w.Write(jpg); err != nil {
		return err
	}
	if _, err := w.Write([]byte("\r\n")); err != nil {
		return err
	}
	// Flushed per frame. Anything held waiting for a full buffer is a tile
	// that stays blank, which is indistinguishable from the view not working.
	if flusher != nil {
		flusher.Flush()
	}
	return nil
}