bbc-on-ice/main.go

307 lines
7.4 KiB
Go

package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"io/ioutil"
"log"
"net"
"net/http"
"os"
"path"
"regexp"
"strconv"
"strings"
"syscall"
"time"
"unicode/utf8"
)
type meta struct {
title string // What's playing right now
timeout uint // Timeout for the next poll in ms
}
// Retrieve and decode metadata information from an independent webservice
func getMeta(name string) (*meta, error) {
const metaBaseURI = "http://polling.bbc.co.uk/radio/nhppolling/"
resp, err := http.Get(metaBaseURI + name)
if resp != nil {
defer resp.Body.Close()
}
if err != nil {
return nil, err
}
b, err := ioutil.ReadAll(resp.Body)
if len(b) < 2 {
// There needs to be an enclosing () pair
return nil, errors.New("invalid metadata response")
}
// TODO: also retrieve richtracks/is_now_playing, see example file
type broadcast struct {
Title string // Title of the broadcast
Percentage int // How far we're in
}
var v struct {
Packages struct {
OnAir struct {
Broadcasts []broadcast
BroadcastNowIndex uint
} `json:"on-air"`
}
Timeouts struct {
PollingTimeout uint `json:"polling_timeout"`
}
}
err = json.Unmarshal(b[1:len(b)-1], &v)
if err != nil {
return nil, errors.New("invalid metadata response")
}
onAir := v.Packages.OnAir
if onAir.BroadcastNowIndex >= uint(len(onAir.Broadcasts)) {
return nil, errors.New("no active broadcast")
}
return &meta{
timeout: v.Timeouts.PollingTimeout,
title: onAir.Broadcasts[onAir.BroadcastNowIndex].Title,
}, nil
}
// Resolve an M3U8 playlist to the first link that seems to be playable
func resolveM3U8(target string) (out []string, err error) {
resp, err := http.Get(target)
if resp != nil {
defer resp.Body.Close()
}
if err != nil {
return nil, err
}
b, err := ioutil.ReadAll(resp.Body)
if !utf8.Valid(b) {
return nil, errors.New("invalid UTF-8")
}
lines := strings.Split(string(b), "\n")
for _, line := range lines {
if strings.HasPrefix(line, "#") {
continue
}
if !strings.Contains(line, "/") {
// Seems to be a relative link, let's make it absolute
dir, _ := path.Split(target)
line = dir + line
}
if strings.HasSuffix(line, "m3u8") {
// The playlist seems to recurse, and so do we
return resolveM3U8(line)
}
out = append(out, line)
}
return out, nil
}
func metaProc(ctx context.Context, name string, out chan<- string) {
defer close(out)
var current, last string
var interval time.Duration
for {
meta, err := getMeta(name)
if err != nil {
current = "Error: " + err.Error()
interval = 30 * time.Second
} else {
current = meta.title
interval = time.Duration(meta.timeout)
}
if current != last {
// TODO: see https://blog.golang.org/pipelines
// find out if we can do this better
select {
case out <- current:
case <-ctx.Done():
return
}
last = current
}
select {
case <-time.After(time.Duration(interval) * time.Millisecond):
case <-ctx.Done():
return
}
}
}
var pathRE = regexp.MustCompile(`^/(.*?)/(.*?)/(.*?)$`)
func proxy(w http.ResponseWriter, req *http.Request) {
const targetURI = "http://a.files.bbci.co.uk/media/live/manifesto/" +
"audio/simulcast/hls/%s/%s/ak/%s.m3u8"
const metaint = 1 << 16
m := pathRE.FindStringSubmatch(req.URL.Path)
if m == nil {
http.NotFound(w, req)
return
}
hijacker, ok := w.(http.Hijacker)
if !ok {
// We're not using TLS where HTTP/2 could have caused this
panic("cannot hijack connection")
}
// E.g. `nonuk`, `sbr_low` `bbc_radio_one`, or `uk`, `sbr_high`, `bbc_1xtra`
region, quality, name := m[1], m[2], m[3]
// This validates the params as a side-effect
target, err := resolveM3U8(fmt.Sprintf(targetURI, region, quality, name))
if err == nil && len(target) == 0 {
err = errors.New("cannot resolve playlist")
}
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
wantMeta := false
if icyMeta, ok := req.Header["Icy-MetaData"]; ok {
wantMeta = len(icyMeta) == 1 && icyMeta[0] == "1"
}
resp, err := http.Get(target[0])
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
conn, bufrw, err := hijacker.Hijack()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
defer conn.Close()
// TODO: retrieve some general information from somewhere?
// There's nothing interesting in the playlist files.
fmt.Fprintf(bufrw, "ICY 200 OK\r\n")
fmt.Fprintf(bufrw, "icy-name:%s\r\n", name)
// BBC marks this as a video type, maybe just force audio/mpeg
fmt.Fprintf(bufrw, "content-type:%s\r\n", resp.Header["Content-Type"][0])
fmt.Fprintf(bufrw, "icy-pub:%d\r\n", 0)
if wantMeta {
fmt.Fprintf(bufrw, "icy-metaint: %d\r\n", metaint)
}
fmt.Fprintf(bufrw, "\r\n")
metaChan := make(chan string)
go metaProc(req.Context(), name, metaChan)
// TODO: move to a normal function
// FIXME: this will load a few seconds (one URL) and die
// - we can either try to implement this and hope for the best
// https://tools.ietf.org/html/draft-pantos-http-live-streaming-20
// then like https://github.com/kz26/gohls/blob/master/main.go
// - or we can become more of a proxy, which complicates ICY
chunkChan := make(chan []byte)
go func() {
defer resp.Body.Close()
defer close(chunkChan)
for {
chunk := make([]byte, metaint)
n, err := io.ReadFull(resp.Body, chunk)
chunkChan <- chunk[:n]
if err != nil {
return
}
select {
default:
case <-req.Context().Done():
return
}
}
}()
var queuedMeta []byte
for {
select {
case title := <-metaChan:
queuedMeta = []byte(fmt.Sprintf("StreamTitle='%s'", title))
case chunk, ok := <-chunkChan:
if !ok {
return
}
if wantMeta {
var meta [1 + 16*255]byte
meta[0] = byte((copy(meta[1:], queuedMeta) + 15) / 16)
chunk = append(chunk, meta[:1+int(meta[0])*16]...)
queuedMeta = nil
}
if _, err := bufrw.Write(chunk); err != nil {
return
}
if err := bufrw.Flush(); err != nil {
return
}
}
}
}
// https://www.freedesktop.org/software/systemd/man/sd_listen_fds.html
func socketActivationListener() net.Listener {
pid, err := strconv.Atoi(os.Getenv("LISTEN_PID"))
if err != nil || pid != os.Getpid() {
return nil
}
nfds, err := strconv.Atoi(os.Getenv("LISTEN_FDS"))
if err != nil || nfds == 0 {
return nil
} else if nfds > 1 {
log.Fatalln("not supporting more than one listening socket")
}
const firstListenFd = 3
syscall.CloseOnExec(firstListenFd)
ln, err := net.FileListener(os.NewFile(firstListenFd, "socket activation"))
if err != nil {
log.Fatalln(err)
}
return ln
}
// Had to copy this from Server.ListenAndServe()
type tcpKeepAliveListener struct{ *net.TCPListener }
func (ln tcpKeepAliveListener) Accept() (c net.Conn, err error) {
tc, err := ln.AcceptTCP()
if err != nil {
return
}
_ = tc.SetKeepAlive(true)
_ = tc.SetKeepAlivePeriod(3 * time.Minute)
return tc, nil
}
func main() {
listenAddr := ":8000"
if len(os.Args) == 2 {
listenAddr = os.Args[1]
}
var listener net.Listener
if ln := socketActivationListener(); listener != nil {
// Keepalives can be set in the systemd unit, see systemd.socket(5)
listener = ln
} else if ln, err := net.Listen("tcp", listenAddr); err != nil {
log.Fatalln(err)
} else {
listener = tcpKeepAliveListener{ln.(*net.TCPListener)}
}
http.HandleFunc("/", proxy)
// We don't need to clean up properly since we store no data
log.Fatalln(http.Serve(listener, nil))
}