如何延迟直播？

Question

I'm trying to build a service in Go which delays a live stream (socketio/signalR) for ~7 minutes. It should also allow a non-delayed stream. So the Go service should have something like a buffer or a queue that forces data to wait for the specified duration before it is allowed to be consumed. How would you do something like this in Go? Would the delayed stream be a seperate goroutine? What data structure should be used to delay the data?

My current idea would be to use the time package to wait/tick for 7 minutes before the data is allowed to be consumed but this blocking behavior might not be optimal in this scenario.

Here is some code to explain what I'm trying to do. FakeStream is a mock function that simulates the live streamed data I'm getting from the external service.

package main

import (
    "fmt"
    "time"
)

func DelayStream(input chan string, output chan string, delay string) {

    // not working for some reason
    // delayDuration, _ := time.ParseDuration(delay)
    // fmt.Println(delayDuration.Seconds())

    if delay == "5s" {
        fmt.Println("sleeping")
        time.Sleep(5 * time.Second)
    }
    data := <-input
    output <- data
}

func FakeStream(live chan string) {

    ticks := time.Tick(2 * time.Second)
    for now := range ticks {
        live <- fmt.Sprintf("%v", now.Format(time.UnixDate))
    }
}

func main() {
    liveData := make(chan string)
    delayedData := make(chan string)

    go FakeStream(liveData)
    go DelayStream(liveData, delayedData, "5s")

    for {
        select {
        case live := <-liveData:
            fmt.Println("live: ", live)
        case delayed := <-delayedData:
            fmt.Println("delayed: ", delayed)
        }
    }
}

For some reason the delayed channel only outputs once and it doesn't output the expected data. It should output the first thing in the live channel but it doesn't.

Answer 1

You need a buffer of sufficient size. For simple cases, a buffered Go channel could work.

Ask yourself - how much data is there to store during this delay - you should have a reasonable upper cap. For example if your stream delivers up to N packets per second, then to delay by 7 minutes you'll need to store 420N packets.

Ask yourself - what happens if more data than expected arrives during the delay window? You can throw the new data away, or throw the old data away, or just block the input stream. Which of these are feasible for your scenario? Each results in a slightly different solution.

Ask yourself - how is the delay computed? From the moment the stream is created? From the moment each packet arrives? Is the delay for each packet separately, or only for the first packet in the stream?

You'll need to considerably narrow down the design choices here in order to develop some sample code.

For some subset of these design choices, here's a simple way to add delay between channels for each message:

---

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    // in is a channel of strings with a buffer size of 10
    in := make(chan string, 10)

    // out is an unbuffered channel
    out := make(chan string)

    // this goroutine forwards messages from in to out, ading a delay
    // to each message.
    const delay = 3 * time.Second
    go func() {
        for msg := range in {
            time.Sleep(delay)
            out <- msg
        }
        close(out)
    }()

    var wg sync.WaitGroup
    wg.Add(1)
    // this goroutine drains the out channel
    go func() {
        for msg := range out {
            fmt.Printf("Got '%s' at time %s
", msg, time.Now().Format(time.Stamp))
        }
        wg.Done()
    }()

    // Send some messages into the in channel
    fmt.Printf("Sending '%s' at time %s
", "joe", time.Now().Format(time.Stamp))
    in <- "joe"

    time.Sleep(2 * time.Second)
    fmt.Printf("Sending '%s' at time %s
", "hello", time.Now().Format(time.Stamp))
    in <- "hello"

    time.Sleep(4 * time.Second)
    fmt.Printf("Sending '%s' at time %s
", "bye", time.Now().Format(time.Stamp))
    in <- "bye"
    close(in)

    wg.Wait()
}