Golang Video stream Input/Output multiplexing

Question

I wrote a camera client that will read a video stream and write to a player for playback.

I have some basic keyboard controls for recording the stream.

I coded the client with a single video stream in mind.

I refactored the code so I can have playback from multiple streams but I am having trouble with the keyboard controls.

I am really struggling with figuring out a model for properly communicating to the client from which camera it should record.

Below the code.

All of it works, but with the current model the client records from a random camera.

main.go

package main

import (
    "apps/piClient/auth"
    "apps/piClient/client"
    "bufio"
    "fmt"
    "log"
    "os"
    "os/exec"
    "sync"
    "syscall"

    "github.com/BurntSushi/toml"
)

func main() {
var config client.Configuration

_, err := toml.DecodeFile("config.toml", &config)
if err != nil {
    log.Fatal(err)
}

var wg sync.WaitGroup
wg.Add(1) //<- Extra points if you can tell me how to handle
//my goroutines. All are never ending for loops
record := make(chan string)

for id, camera := range config.Cameras {
    h264 := make(chan client.Frame)

    num := fmt.Sprintf("%v", id)
    stream := "stream" + num
    fromClientPort := ":808" + num

    err = syscall.Mkfifo(stream, 0775)
    if err != nil {
        log.Println(err)
    }

    pipe, err := os.OpenFile(stream, os.O_RDWR, os.ModeNamedPipe)
    if err != nil {
        log.Fatal(err)
    }

    w := bufio.NewWriter(pipe)

    conn, err := auth.Login(camera.Address, camera.Password, fromClientPort, camera.Retries)
    if err != nil {
        log.Println(err)
        continue
    }

    defer conn.Close()

    go client.Read(conn, h264)
    go client.Write(w, h264, record)

    go playerRun(stream)

    fmt.Printf("Press 'r + return' to start recording.\nPress 's + return' to stop recording.\n")

    go func() {
        reader := bufio.NewReader(os.Stdin)
        for {
            char, _, err := reader.ReadRune()
            if err != nil {
                log.Fatal(err)
            }

            switch char {
            case 'r':
                record <- "start"
                char = '\x00'
            case 's':
                record <- "stop"
                char = '\x00'
            }
        }
    }()
}
wg.Wait()
}

func playerRun(stream string) {
    cmd := exec.Command("./play", stream)
    //cmd.Stderr = os.Stdout
    //cmd.Stdout = os.Stdout

    if err := cmd.Start(); err != nil {
        log.Fatal(err)
    }
}

client.go

package client

import (
    "bufio"
    "fmt"
    "log"
    "net"
    "os"
    "strings"
    "time"
)

type Configuration struct {
    Cameras []Camera
}

type Camera struct {
    Address  string
    Retries  int
    Password string
}

type state bool
type Frame []byte

func Read(conn *net.UDPConn, output chan<- Frame) {
    for {
        buf := make([]byte, 65507)
        n, _, err := conn.ReadFromUDP(buf)
        if err != nil {
            log.Fatal(err)
        }

        if n > 0 {
            output <- buf[0:n]
        }
    }
}

func Write(w *bufio.Writer, input <-chan Frame, record <-chan string) {
    var recording state
    var fw *bufio.Writer
    var file *os.File

    for {
        select {
        case action := <-record:
            switch action {
            case "start":
                if !recording {
                    fw, file = newFileWriter()
                    recording = true
                    fmt.Println("recording...")
                } else {
                    fmt.Println("client is already recording")
                }
            case "stop":
                if recording {
                    fw.Flush()
                    file.Close()
                    recording = false
                    fmt.Println("recording stopped")
                } else {
                    fmt.Println("client is not recording")
                }
            }
        case h264 := <-input:
            _, err := w.Write(h264)
            if err != nil {
                log.Fatal(err)
            }
            if recording {
                _, err = fw.Write(h264)
                if err != nil {
                    log.Fatal(err)
                }
            }
        }
    }
}

func newFileWriter() (*bufio.Writer, *os.File) {
    now := time.Now()
    dir := "video/"
    fname := fmt.Sprintf("%v.h264", now)
    fname = strings.Replace(fname, " ", "-", -1)
    fname = strings.Replace(fname, ":", "-", -1)

    file, err := os.Create(dir + fname)
    if err != nil {
        log.Fatal(err)
    }

    fw := bufio.NewWriter(file)
    return fw, file
}

Answer 1

1. I got rid of WaitGroup and instead used a channel that all goroutines listen to in case there is an explicit kill.
2. You can specify which camera to record from cmd line itself

package main  

import (
    "apps/piClient/auth"
    "apps/piClient/client"
    "bufio"
    "fmt"
    "log"
    "os"
    "os/exec"
    "syscall"
    "strconv"
    "strings"
    "os/signal"
    "time"

    "github.com/BurntSushi/toml"
)

func main() {
    var config client.Configuration

    _, err := toml.DecodeFile("config.toml", &config)
    if err != nil {
        log.Fatal(err)
    }

    done := make(chan bool)
    recorder := make(map[int]chan<- string)

    for id, camera := range config.Cameras {
        stream := fmt.Sprintf("stream%d", id)
        fromClientPort := fmt.Sprintf(":808%d", id)

        err = syscall.Mkfifo(stream, 0775)
        if err != nil {
            log.Println(err)
            continue
        }

        pipe, err := os.OpenFile(stream, os.O_RDWR, os.ModeNamedPipe)
        if err != nil {
            log.Println(err)
            continue
        }

        w := bufio.NewWriter(pipe)

        conn, err := auth.Login(camera.Address, camera.Password, fromClientPort, camera.Retries)
        if err != nil {
            log.Println(err)
            continue
        }
        h264 := make(chan client.Frame)
        c := client.NewClient(id, camera)
        record := make(chan string)
        go c.Read(conn, h264, done)
        go c.Write(w, h264, record, done)
        go playerRun(stream)
        recorder[id] = record
        fmt.Printf(
            "Press 'r%d + return' to start recording.\nPress 's%d + return' to stop recording.\n",
            id, id)
    }

    go func(done <-chan bool) {
        reader := bufio.NewReader(os.Stdin)
        for {
            select {
            case <-done:
                return
            default:
                line, _, _ := reader.ReadLine()
                data := strings.Split(string(line), "")
                id, err := strconv.Atoi(data[1])
                if err != nil {
                    log.Println(err)
                    continue
                }

                recordChan := recorder[id]
                switch data[0] {
                case 'r':
                    recordChan <- "start"
                case 's':
                    recordChan <- "stop"
                }
            }
        }
    }()

    c := make(chan os.Signal, 1)
    signal.Notify(c, os.Interrupt)
    for range c {
        close(done)
        break
    }
    // wait for clean up
    time.Sleep(2 * time.Second)
}

func playerRun(stream string) {
    cmd := exec.Command("./play", stream)
    //cmd.Stderr = os.Stdout
    //cmd.Stdout = os.Stdout

    if err := cmd.Start(); err != nil {
        log.Fatal(err)
    }
}

package client

import (
    "bufio"
    "fmt"
    "log"
    "net"
    "os"
    "strings"
    "time"
)

type Client struct {
    ID        int
    Camera    Camera
    recording state
}

func NewClient(id int, c Camera) *Client {
    return &Client{
        ID:     id,
        Camera: c,
    }
}

type Configuration struct {
    Cameras []Camera
}

type Camera struct {
    Address  string
    Retries  int
    Password string
}

type state bool
type Frame []byte

func (client *Client) Read(conn *net.UDPConn, output chan<- Frame, done <-chan bool) {
    for {
        select {
        case <-done:
            conn.Close()
            return
        default:
            buf := make([]byte, 65507)
            n, _, err := conn.ReadFromUDP(buf)
            if err != nil {
                conn.Close()
                log.Fatalf("Client %d failed with error %v", client.ID, err)
            }

            if n > 0 {
                output <- buf[0:n]
            }
        }
    }
}

func (client *Client) Write(w *bufio.Writer, input <-chan Frame, record <-chan string, done <-chan bool) {
    var fw *bufio.Writer
    var file *os.File

    for {
        select {
        case <-done:
            w.Flush()
            if client.recording {
                client.recording = false
                fw.Flush()
                file.Close()
            }
            return
        case action := <-record:
            switch action {
            case "start":
                if !client.recording {
                    fw, file = newFileWriter()
                    client.recording = true
                    fmt.Println("recording...")
                } else {
                    fmt.Println("client is already recording")
                }
            case "stop":
                if client.recording {
                    fw.Flush()
                    file.Close()
                    client.recording = false
                    fmt.Println("recording stopped")
                } else {
                    fmt.Println("client is not recording")
                }
            }
        case h264 := <-input:
            _, err := w.Write(h264)
            if err != nil {
                log.Fatal(err)
            }
            if client.recording {
                _, err = fw.Write(h264)
                if err != nil {
                    log.Fatal(err)
                }
            }
        }
    }
}

func newFileWriter() (*bufio.Writer, *os.File) {
    now := time.Now()
    dir := "video/"
    fname := fmt.Sprintf("%v.h264", now)
    fname = strings.Replace(fname, " ", "-", -1)
    fname = strings.Replace(fname, ":", "-", -1)

    file, err := os.Create(dir + fname)
    if err != nil {
        log.Fatal(err)
    }

    fw := bufio.NewWriter(file)
    return fw, file
}