Ggtcf

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.

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 %sn", msg, time.Now().Format(time.Stamp))
 
 wg.Done()
 ()

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

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

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

 wg.Wait()