diff options
| -rw-r--r-- | .gitignore | 4 | ||||
| -rw-r--r-- | Makefile | 45 | ||||
| -rw-r--r-- | cmd/dolmetschctld/dolmetschctld.go | 8 | ||||
| -rw-r--r-- | cmd/dolmetschctld/statemachine.go | 40 | ||||
| -rw-r--r-- | cmd/dolmetschctld/telnet.go | 2 | ||||
| -rw-r--r-- | go.mod | 2 | ||||
| -rw-r--r-- | go.sum | 4 | ||||
| -rw-r--r-- | pkg/mixer/config.go | 4 | ||||
| -rw-r--r-- | pkg/mixer/mixer.go | 315 |
9 files changed, 249 insertions, 175 deletions
@@ -8,8 +8,8 @@ app/*.o app/*.d app/*.d.* app/config.h -cmd/dolmetschctl/dolmetschctl -cmd/dolmetschctld/dolmetschctld +dolmetschctl +dolmetschctld firmware/*.a firmware/*.o firmware/*.d diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..a557e34 --- /dev/null +++ b/Makefile @@ -0,0 +1,45 @@ +## +## dolmetschctl +## +## +## Copyright (C) 2019 Christian Pointner <equinox@spreadspace.org> +## +## This file is part of dolmetschctl. +## +## dolmetschctl is free software: you can redistribute it and/or modify +## it under the terms of the GNU General Public License as published by +## the Free Software Foundation, either version 3 of the License, or +## any later version. +## +## dolmetschctl is distributed in the hope that it will be useful, +## but WITHOUT ANY WARRANTY; without even the implied warranty of +## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +## GNU General Public License for more details. +## +## You should have received a copy of the GNU General Public License +## along with dolmetschctl. If not, see <http://www.gnu.org/licenses/>. +## + +GOCMD := go +ifdef GOROOT +GOCMD = $(GOROOT)/bin/go +endif + +all: dolmetschctl dolmetschctld +.PHONY: vet format dolmetschctl dolmetschctld clean + +vet: + $(GOCMD) vet ./... + +format: + $(GOCMD) fmt ./... + +dolmetschctl: + $(GOCMD) build -o dolmetschctl ./cmd/dolmetschctl + +dolmetschctld: + $(GOCMD) build -o dolmetschctld ./cmd/dolmetschctld + +clean: + rm -f dolmetschctl + rm -f dolmetschctld diff --git a/cmd/dolmetschctld/dolmetschctld.go b/cmd/dolmetschctld/dolmetschctld.go index 38f71ad..3242dde 100644 --- a/cmd/dolmetschctld/dolmetschctld.go +++ b/cmd/dolmetschctld/dolmetschctld.go @@ -44,14 +44,14 @@ func main() { // ************************************* // * initialize core - m, err := mixer.NewMixer(mixer.Config{DevIn: "ESI MIDIMATE eX MIDI 2", DevOut: "ESI MIDIMATE eX MIDI 1"}) + m, err := mixer.NewMixer(mixer.Config{Host: "192.168.48.232"}) if err != nil { log.Printf("Error opening the mixer: %v", err) os.Exit(1) } defer m.Shutdown() - log.Printf("successfully connected to mixer!") + log.Printf("successfully connected to mixer: %s", m.String()) if err = m.Init(); err != nil { log.Printf("Error initializeing the mixer: %v", err) @@ -67,8 +67,8 @@ func main() { log.Printf("state machine successfully initialized!") // TODO: make this configurable - addLanguage(sm, "en", 0x20, 0x24) - addLanguage(sm, "de", 0x22, 0x25) + addLanguage(sm, "en", "auxin/01", "auxin/05") + addLanguage(sm, "de", "auxin/03", "auxin/06") sm.Start() defer sm.Shutdown() diff --git a/cmd/dolmetschctld/statemachine.go b/cmd/dolmetschctld/statemachine.go index 3e92f00..1abb5fd 100644 --- a/cmd/dolmetschctld/statemachine.go +++ b/cmd/dolmetschctld/statemachine.go @@ -42,6 +42,7 @@ type MixerChannel struct { num mixer.Channel target MixerChannelState current MixerChannelState + real MixerChannelState } type MixerChannels struct { @@ -133,7 +134,7 @@ func (sm *StateMachine) publishState() { sm.stateSubscribers.Remove(entry) default: select { - case sub.publish <- types.FullState{sm.state, sm.original2InterpreterRatio, sm.language}: + case sub.publish <- types.FullState{State: sm.state, Ratio: sm.original2InterpreterRatio, Language: sm.language}: default: // subscriber is not responding... log.Printf("statemachine: removing subscriber '%v', because it is not responding", sub.publish) @@ -169,21 +170,16 @@ func (sm *StateMachine) handleMixerEvent(ev mixer.Event) { switch ev.Type { case mixer.EventFaderChange: - mc.current.level = ev.Level + mc.real.level = ev.Level case mixer.EventMute: - mc.current.mute = ev.Mute + mc.real.mute = ev.Mute } } // make sure that our state and the mixer are in sync func (sm *StateMachine) initMixer() { for _, mcs := range sm.languages { - sm.mixer.SetLevel(mcs.original.num, mixer.FaderLevel0db-1) - sm.mixer.SetLevel(mcs.original.num, mixer.FaderLevel0db) mcs.original.target.level = mixer.FaderLevel0db - - sm.mixer.SetLevel(mcs.interpreter.num, mixer.FaderLevelOff+1) - sm.mixer.SetLevel(mcs.interpreter.num, mixer.FaderLevelOff) mcs.interpreter.target.level = mixer.FaderLevelOff } sm.language = "" @@ -209,7 +205,6 @@ func (sm *StateMachine) setLanguage(l types.Language) error { } } sm.language = l - log.Printf("new target language: '%s'", sm.language) return nil } @@ -218,17 +213,16 @@ func (sm *StateMachine) setOriginal2InterpreterRatio(r float32) error { return fmt.Errorf("original-interpreter ratio '%1.3f' is invalid, must be between 0.0 and 1.0", r) } sm.original2InterpreterRatio = r - log.Printf("new original-interpreter ratio: '%1.3f'", r) return nil } func calcNextLevel(target, current mixer.FaderLevel) mixer.FaderLevel { next := target if current != mixer.FaderLevelUnknown { - if next > current { - next = current + 1 - } else { - next = current - 1 + if next > (current + mixer.FaderLevelIncrement/2) { + next = current + mixer.FaderLevelIncrement + } else if next < (current - mixer.FaderLevelIncrement/2) { + next = current - mixer.FaderLevelIncrement } } return next @@ -236,7 +230,7 @@ func calcNextLevel(target, current mixer.FaderLevel) mixer.FaderLevel { func (sm *StateMachine) reconcile(ticker bool) { for lang, mcs := range sm.languages { - if sm.language == "" || lang == sm.language || mcs.interpreter.current.mute == mixer.MuteMuted { + if sm.language == "" || lang == sm.language { mcs.original.target.level = mixer.FaderLevel0db mcs.interpreter.target.level = mixer.FaderLevelOff } else { @@ -251,14 +245,19 @@ func (sm *StateMachine) reconcile(ticker bool) { if sm.state != types.StateSettled && !ticker { return } + sm.state = types.StateSettled for _, mcs := range sm.languages { if mcs.original.target.level != mcs.original.current.level { - sm.mixer.SetLevel(mcs.original.num, calcNextLevel(mcs.original.target.level, mcs.original.current.level)) + nextLevel := calcNextLevel(mcs.original.target.level, mcs.original.current.level) + sm.mixer.SetLevel(mcs.original.num, nextLevel) + mcs.original.current.level = nextLevel sm.state = types.StateSettling } if mcs.interpreter.target.level != mcs.interpreter.current.level { - sm.mixer.SetLevel(mcs.interpreter.num, calcNextLevel(mcs.interpreter.target.level, mcs.interpreter.current.level)) + nextLevel := calcNextLevel(mcs.interpreter.target.level, mcs.interpreter.current.level) + sm.mixer.SetLevel(mcs.interpreter.num, nextLevel) + mcs.interpreter.current.level = nextLevel sm.state = types.StateSettling } } @@ -292,7 +291,7 @@ func (sm *StateMachine) run() { case req := <-sm.getOriginal2InterpreterRatioCh: req.resCh <- sm.original2InterpreterRatio case req := <-sm.getStateCh: - req.resCh <- getStateRes{types.FullState{sm.state, sm.original2InterpreterRatio, sm.language}} + req.resCh <- getStateRes{types.FullState{State: sm.state, Ratio: sm.original2InterpreterRatio, Language: sm.language}} case ev := <-sm.mixerEventCh: sm.handleMixerEvent(ev) sm.reconcile(false) @@ -301,6 +300,7 @@ func (sm *StateMachine) run() { } if oldState != sm.state || oldOriginal2InterpreterRadio != sm.original2InterpreterRatio || oldLanguage != sm.language { + log.Printf("update: lang=%s, ratio=%1.3f (%s)", sm.language, sm.original2InterpreterRatio, sm.state) sm.publishState() } } @@ -344,8 +344,8 @@ func (sm *StateMachine) AddLanguage(name types.Language, original, interpreter m } unknown := MixerChannelState{level: mixer.FaderLevelUnknown, mute: mixer.MuteUnknown} - chOriginal := MixerChannel{num: original, current: unknown, target: unknown} - chInterpreter := MixerChannel{num: interpreter, current: unknown, target: unknown} + chOriginal := MixerChannel{num: original, target: unknown, current: unknown, real: unknown} + chInterpreter := MixerChannel{num: interpreter, target: unknown, current: unknown, real: unknown} sm.languages[name] = &MixerChannels{chOriginal, chInterpreter} sm.channel2lang[original] = name diff --git a/cmd/dolmetschctld/telnet.go b/cmd/dolmetschctld/telnet.go index 5618ebe..55c3bf0 100644 --- a/cmd/dolmetschctld/telnet.go +++ b/cmd/dolmetschctld/telnet.go @@ -78,7 +78,7 @@ func telnetCmdState(c *telgo.Client, args []string, sm *StateMachine) bool { } state, ratio, lang := sm.GetState() c.Sayln("current state: %s", state) - c.Sayln("current ratio: %s", ratio) + c.Sayln("current ratio: %f", ratio) c.Sayln("current language: %s", lang) return false } @@ -4,7 +4,9 @@ go 1.13 require ( github.com/gorilla/websocket v1.4.1 + github.com/imdario/go-ulid v0.0.0-20180116185620-aeb52bf96595 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/scgolang/midi v0.5.0 + github.com/scgolang/osc v0.11.1 github.com/spreadspace/telgo v0.0.0-20170609015223-7277b0d8090e ) @@ -1,8 +1,12 @@ github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM= github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/imdario/go-ulid v0.0.0-20180116185620-aeb52bf96595 h1:8MKHx/6AMMFGslqvr37RF7zktr3eJmY1z2FKdq3Zo/o= +github.com/imdario/go-ulid v0.0.0-20180116185620-aeb52bf96595/go.mod h1:ugPCasYVpR6Cf8xlF0vkZdVKntj7zTgo9pLR4Si7Boo= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/scgolang/midi v0.5.0 h1:HsSMHIbgCAZlmjloOgRZeLQgBXi8eWXw3FZJzm3eEkE= github.com/scgolang/midi v0.5.0/go.mod h1:ur2Ay6ad9oWPNFuGq4oQ6u75uZIjZfDdhozbGHnTcfQ= +github.com/scgolang/osc v0.11.1 h1:o2+nXrQrlyEAoFcgZ2zk6p5iI6ht+NgiSKaGQBpvWbU= +github.com/scgolang/osc v0.11.1/go.mod h1:fu5QITvJ5w2pzKXJBmyVTF89ZycPN4bS4cOHJErpR2A= github.com/spreadspace/telgo v0.0.0-20170609015223-7277b0d8090e h1:lhFaE8l5AfSEzH4OW+9BUAo+BuYLe21zuHwWX1YEZHw= github.com/spreadspace/telgo v0.0.0-20170609015223-7277b0d8090e/go.mod h1:kvvhiAI5nvwI4SZrVH9vRa3MTjkcYm3tElmmh+bYuEI= diff --git a/pkg/mixer/config.go b/pkg/mixer/config.go index bc0a292..e25ba2e 100644 --- a/pkg/mixer/config.go +++ b/pkg/mixer/config.go @@ -23,6 +23,6 @@ package mixer type Config struct { - DevIn string `json:"dev-in" yaml:"dev-in" toml:"dev-in"` - DevOut string `json:"dev-out" yaml:"dev-out" toml:"dev-out"` + Host string `json:"host" yaml:"host" toml:"host"` + Port string `json:"port" yaml:"port" toml:"port"` } diff --git a/pkg/mixer/mixer.go b/pkg/mixer/mixer.go index afa1534..dbc4dae 100644 --- a/pkg/mixer/mixer.go +++ b/pkg/mixer/mixer.go @@ -27,20 +27,19 @@ import ( "errors" "fmt" "log" + "net" "strings" "sync" "time" - "github.com/scgolang/midi" + "github.com/scgolang/osc" ) -// TODO: make this configurable const ( - CC_MUTE = byte(0xB1) - CC_FADER = byte(0xB0) + OSCSubscriptionTimeFactor = 0 ) -type Channel uint8 +type Channel string type EventType int const ( @@ -59,21 +58,21 @@ func (et EventType) String() string { } } -type FaderLevel uint8 +type FaderLevel float32 -// TODO: make the values configurable const ( - FaderLevelUnknown = FaderLevel(0xFF) - FaderLevelMax = FaderLevel(0x7F) - FaderLevel0db = FaderLevel(0x60) - FaderLevelOff = FaderLevel(0x00) + FaderLevelUnknown = FaderLevel(-1.0) + FaderLevelMax = FaderLevel(1.00) + FaderLevel0db = FaderLevel(0.75) + FaderLevelOff = FaderLevel(0.00) + FaderLevelIncrement = FaderLevel(0.01) ) func (fl FaderLevel) String() string { if fl > FaderLevelMax { return "unknown" } - val := fmt.Sprintf("%3d", fl) + val := fmt.Sprintf("%3f", fl) switch fl { case FaderLevelMax: return val + " (max)" @@ -86,7 +85,7 @@ func (fl FaderLevel) String() string { } } -type Mute int8 +type Mute int const ( MuteUnknown = Mute(-1) @@ -113,7 +112,7 @@ type Event struct { } func (e Event) String() string { - return fmt.Sprintf("Event(%s) for channel %d: level=%s, muted=%s", e.Type, e.Channel, e.Level, e.Mute) + return fmt.Sprintf("Event(%s) for channel '%s': level=%s, muted=%s", e.Type, e.Channel, e.Level, e.Mute) } type subscriber struct { @@ -121,96 +120,89 @@ type subscriber struct { unsubscribe <-chan struct{} } +type MixerInfo struct { + ServerVersion string + ServerName string + ConsoleModel string + ConsoleVersion string +} + type Mixer struct { config Config - devIn *midi.Device - devOutLock sync.Mutex - devOut *midi.Device + client *osc.UDPConn + info MixerInfo subscribersLock sync.Mutex subscribers map[Channel]*list.List } -func openDevice(devices []*midi.Device, prefix string) (d *midi.Device, err error) { - for _, device := range devices { - if strings.HasPrefix(device.Name, prefix) { - d = device - } - } - if d == nil { - return nil, errors.New("could not find device with prefix: " + prefix) - } - return d, d.Open() +func waitForInfo(client *osc.UDPConn, infoCh chan<- MixerInfo, errCh chan<- error) { + err := client.Serve(1, osc.PatternMatching{ + "/info": osc.Method(func(msg osc.Message) (err error) { + mi := MixerInfo{} + if len(msg.Arguments) != 4 { + return errors.New("invalid number of arguments in /info response") + } + if mi.ServerVersion, err = msg.Arguments[0].ReadString(); err != nil { + return + } + if mi.ServerName, err = msg.Arguments[1].ReadString(); err != nil { + return + } + if mi.ConsoleModel, err = msg.Arguments[2].ReadString(); err != nil { + return + } + if mi.ConsoleVersion, err = msg.Arguments[3].ReadString(); err != nil { + return + } + + infoCh <- mi + return nil + }), + }) + errCh <- err + + close(infoCh) + close(errCh) } func NewMixer(c Config) (*Mixer, error) { - devices, err := midi.Devices() + if c.Port == "" { + c.Port = "10023" + } + m := &Mixer{config: c} + + server, err := net.ResolveUDPAddr("udp", c.Host+":"+c.Port) + if err != nil { + return nil, err + } + m.client, err = osc.DialUDP("udp", nil, server) if err != nil { return nil, err } - // TODO: add support for devIn == devOut - m := &Mixer{config: c} - if m.devIn, err = openDevice(devices, c.DevIn); err != nil { + if err = m.client.Send(osc.Message{Address: "/info"}); err != nil { return nil, err } - m.devIn.QueueSize = 100 - if m.devOut, err = openDevice(devices, c.DevOut); err != nil { + + infoCh := make(chan MixerInfo) + errCh := make(chan error) + go waitForInfo(m.client, infoCh, errCh) + + select { + case <-time.After(10 * time.Second): + return nil, errors.New("timeout") + case err := <-errCh: return nil, err + case info := <-infoCh: + m.info = info } m.subscribers = make(map[Channel]*list.List) return m, nil } -func (m *Mixer) reopenInput() { - for { - time.Sleep(time.Second) - - devices, err := midi.Devices() - if err != nil { - log.Printf("mixer: error listing midi devices: %v, retrying...", err) - continue - } - - if m.devIn, err = openDevice(devices, m.config.DevIn); err != nil { - log.Printf("mixer: error re-opening midi input device: %v, retrying...", err) - continue - } - - if err = m.Init(); err != nil { - log.Printf("mixer: error re-initializing midi input device: %v, retrying...", err) - - continue - } - - log.Printf("mixer: successfully re-initialized midi input device") - break - } -} - -func (m *Mixer) reopenOutput() { - for { - time.Sleep(time.Second) - - devices, err := midi.Devices() - if err != nil { - log.Printf("mixer: error listing midi devices: %v, retrying...", err) - continue - } - - newOut, err := openDevice(devices, m.config.DevOut) - if err != nil { - log.Printf("mixer: error re-opening midi output device: %v, retrying...", err) - continue - } - - log.Printf("mixer: successfully re-opened midi output device") - - m.devOutLock.Lock() - defer m.devOutLock.Unlock() - m.devOut = newOut - break - } +func (m *Mixer) String() string { + return fmt.Sprintf("Model %s (Firmware-Version: %s)", m.info.ConsoleModel, m.info.ConsoleVersion) } func (m *Mixer) publishEvent(ev Event) { @@ -246,57 +238,86 @@ func (m *Mixer) publishEvent(ev Event) { } } -func (m *Mixer) handleMidiPacket(p midi.Packet) { +type oscDispatcher struct { + mixer *Mixer +} + +func (d oscDispatcher) Dispatch(bundle osc.Bundle, exactMatch bool) error { + // TODO: X32 seems to not use bundles at the momemnt - ingoring them for now + return nil +} + +func (d oscDispatcher) Invoke(msg osc.Message, exactMatch bool) error { + addrParts := strings.Split(strings.TrimPrefix(msg.Address, "/"), "/") + if len(addrParts) < 4 { + return nil + } + if addrParts[2] != "mix" { + return nil + } + if len(msg.Arguments) != 1 { + return nil + } + ev := Event{Level: FaderLevelUnknown, Mute: MuteUnknown} - ev.Channel = Channel(p.Data[1]) - switch p.Data[0] { - case CC_FADER: - ev.Type = EventFaderChange - ev.Level = FaderLevel(p.Data[2]) - case CC_MUTE: + ev.Channel = Channel(addrParts[0] + "/" + addrParts[1]) + switch addrParts[3] { + case "on": ev.Type = EventMute - ev.Mute = MuteUnmuted - if p.Data[2] > 0 { + arg, err := msg.Arguments[0].ReadInt32() + if err != nil { + return err + } + if arg == 0 { ev.Mute = MuteMuted + } else { + ev.Mute = MuteUnmuted + } + case "fader": + ev.Type = EventFaderChange + arg, err := msg.Arguments[0].ReadFloat32() + if err != nil { + return err } + ev.Level = FaderLevel(arg) default: - return + return nil } - m.publishEvent(ev) + + d.mixer.publishEvent(ev) + return nil } func (m *Mixer) Init() error { - ch, err := m.devIn.Packets() - if err != nil { - return err - } + go func() { + for { + dispatcher := oscDispatcher{mixer: m} + err := m.client.Serve(1, dispatcher) + if err != nil { + log.Printf("mixer: failed to dispatch message: %v", err) + time.Sleep(time.Second) + } + } + }() go func() { for { - ps := <-ch - for _, p := range ps { - if p.Err != nil { - log.Printf("mixer: got fatal error from midi input device: %v, trying to reopen it...", p.Err) - m.devIn.Close() - m.devIn = nil - go m.reopenInput() - return - } - m.handleMidiPacket(p) + time.Sleep(5 * time.Second) + err := m.client.Send(osc.Message{Address: "/renew"}) + if err != nil { + log.Printf("mixer: failed to renew subsciptions: %v", err) } } }() + return nil } func (m *Mixer) Shutdown() error { - if m.devIn != nil { - m.devIn.Close() - m.devIn = nil - } - if m.devOut != nil { - m.devOut.Close() - m.devOut = nil + if m.client != nil { + m.client.Send(osc.Message{Address: "/unsubscribe"}) + m.client.Close() + m.client = nil } m.subscribersLock.Lock() @@ -319,59 +340,61 @@ func (m *Mixer) Shutdown() error { return nil } -func (m *Mixer) sendMidiMessage(msg []byte) error { - m.devOutLock.Lock() - defer m.devOutLock.Unlock() - - if m.devOut == nil { - return errors.New("mixer: output device is not ready.") - } - - n, err := m.devOut.Write(msg) - if err != nil { - log.Printf("mixer: got fatal error from midi output device: %v, trying to reopen it...", err) - m.devOut.Close() - m.devOut = nil - go m.reopenOutput() - return err - } - if n != len(msg) { - return errors.New("sending midi message failed: short write") - } - return nil -} - -func (m *Mixer) sendMute(channel byte, value byte) error { - return m.sendMidiMessage([]byte{CC_MUTE, channel, value}) +func (m *Mixer) sendMute(channel Channel, value string) error { + msg := osc.Message{Address: "/" + string(channel) + "/mix/on"} + msg.Arguments = append(msg.Arguments, osc.String(value)) + return m.client.Send(msg) } func (m *Mixer) Mute(ch Channel) error { - return m.sendMute(byte(ch), 0x7F) + return m.sendMute(ch, "OFF") } func (m *Mixer) Unmute(ch Channel) error { - return m.sendMute(byte(ch), 0x00) + return m.sendMute(ch, "ON") } -func (m *Mixer) SetLevel(ch Channel, level FaderLevel) error { +func (m *Mixer) SetLevel(channel Channel, level FaderLevel) error { if level > FaderLevelMax { level = FaderLevelMax } - return m.sendMidiMessage([]byte{CC_FADER, byte(ch), byte(level)}) + msg := osc.Message{Address: "/" + string(channel) + "/mix/fader"} + msg.Arguments = append(msg.Arguments, osc.Float(level)) + return m.client.Send(msg) } -func (m *Mixer) Subscribe(ch Channel, out chan<- Event) chan<- struct{} { +func (m *Mixer) sendSubscribe(channel Channel) (err error) { + msg := osc.Message{Address: "/subscribe"} + msg.Arguments = append(msg.Arguments, osc.String("/"+string(channel)+"/mix/on"), osc.Int(OSCSubscriptionTimeFactor)) + if err = m.client.Send(msg); err != nil { + return + } + + msg = osc.Message{Address: "/subscribe"} + msg.Arguments = append(msg.Arguments, osc.String("/"+string(channel)+"/mix/fader"), osc.Int(OSCSubscriptionTimeFactor)) + if err = m.client.Send(msg); err != nil { + return + } + + return +} + +func (m *Mixer) Subscribe(channel Channel, out chan<- Event) (chan<- struct{}, error) { m.subscribersLock.Lock() defer m.subscribersLock.Unlock() - subs, exists := m.subscribers[ch] + subs, exists := m.subscribers[channel] if !exists { subs = list.New() - m.subscribers[ch] = subs + m.subscribers[channel] = subs + } + + if err := m.sendSubscribe(channel); err != nil { + return nil, err } // log.Printf("mixer: subscribing '%v' to events for channel: %v", out, ch) unsubscribe := make(chan struct{}) subs.PushBack(subscriber{publish: out, unsubscribe: unsubscribe}) - return unsubscribe + return unsubscribe, nil } |