uplink/app/channels/channel.go

package channels

import (
	"log"

	"github.com/cskr/pubsub"

	"git.icedream.tech/icedream/uplink/app/media"
	pubsubutil "git.icedream.tech/icedream/uplink/app/pubsub"
)

type Channel struct {
	InputContainer *media.MediaStreamContainer
	InputStreams   map[string]*media.MediaStream

	OutputContainers map[string]*media.MediaStreamContainer
	OutputStreams    map[string]*media.MediaStream

	Events *pubsub.PubSub

	lastMetadata map[string]string
}

func (channel *Channel) AddInputStream(id string) *media.MediaStream {
	stream := &media.MediaStream{
		PubSubWriter: pubsubutil.NewPubSubWriter(),
	}
	channel.InputStreams[id] = stream
	log.Println("New input stream", id)
	channel.Events.Pub(id, "input_stream")
	return stream
}

func (channel *Channel) AddOutputStream(id string) *media.MediaStream {
	stream := &media.MediaStream{
		PubSubWriter: pubsubutil.NewPubSubWriter(),
	}
	channel.OutputStreams[id] = stream
	log.Println("New output stream", id)
	channel.Events.Pub(id, "output_stream")
	return stream
}

func (channel *Channel) AddOutputContainer(id string) *media.MediaStreamContainer {
	stream := &media.MediaStreamContainer{
		PubSubWriter: pubsubutil.NewPubSubWriter(),
	}
	channel.OutputContainers[id] = stream
	log.Println("New output container", id)
	channel.Events.Pub(id, "output_container")
	return stream
}

func (channel *Channel) SetMetadata(data map[string]string) {
	channel.lastMetadata = data
	channel.Events.Pub(data, "metadata")
}

func (channel *Channel) Metadata() chan map[string]string {
	outC := make(chan map[string]string)
	go func() {
		defer close(outC)
		if channel.lastMetadata != nil {
			outC <- channel.lastMetadata
		}
		c := channel.Events.Sub("metadata")
	forloop:
		for event := range c {
			select {
			case _, _ = <-outC:
				break forloop
			case outC <- event.(map[string]string):
			}
		}
		channel.Events.Unsub(c, "metadata")
	}()
	return outC
}

func NewChannel() *Channel {
	ps := pubsub.New(1)

	inputContainer := pubsubutil.NewPubSubWriterForTopic(ps, "input_container")

	return &Channel{
		InputContainer: &media.MediaStreamContainer{
			PubSubWriter: inputContainer,
		},
		InputStreams: map[string]*media.MediaStream{},

		OutputContainers: map[string]*media.MediaStreamContainer{},
		OutputStreams:    map[string]*media.MediaStream{},

		Events: pubsub.New(1),
	}
}