X-Git-Url: https://git.realraum.at/?p=svn42.git;a=blobdiff_plain;f=go%2Fr3-eventbroker_zmq%2Fmain.go;h=19a335eaf3b80f4a2622d7f8a602edb64e6e3e8e;hp=492e7fb01c31b6231e229b06f77f078f48483ca5;hb=7e85378a44f801ed70cc99e6198c5bcfa9f03d82;hpb=d63674ad22a6c04a1eb0a17d766d189182a5dd70 diff --git a/go/r3-eventbroker_zmq/main.go b/go/r3-eventbroker_zmq/main.go index 492e7fb..19a335e 100644 --- a/go/r3-eventbroker_zmq/main.go +++ b/go/r3-eventbroker_zmq/main.go @@ -3,96 +3,104 @@ package main import ( - "flag" - //~ "time" - pubsub "github.com/tuxychandru/pubsub" - r3events "svn.spreadspace.org/realraum/go.svn/r3events" + "flag" + //~ "time" + pubsub "github.com/tuxychandru/pubsub" + r3events "svn.spreadspace.org/realraum/go.svn/r3events" ) //~ func StringArrayToByteArray(ss []string) [][]byte { - //~ bb := make([][]byte, len(ss)) - //~ for index, s := range(ss) { - //~ bb[index] = []byte(s) - //~ } - //~ return bb +//~ bb := make([][]byte, len(ss)) +//~ for index, s := range(ss) { +//~ bb[index] = []byte(s) +//~ } +//~ return bb //~ } // ---------- Main Code ------------- var ( - doorsub_addr_ string - sensorssub_port_ string - pub_port_ string - keylookup_addr_ string - brain_listen_addr_ string - door_cmd_addr_ string - use_syslog_ bool - enable_debuglog_ bool + doorsub_addr_ string + sensorssub_port_ string + pub_port_ string + keylookup_addr_ string + brain_listen_addr_ string + door_cmd_addr_ string + use_syslog_ bool + enable_debuglog_ bool ) func init() { - flag.StringVar(&door_cmd_addr_, "doorcmdaddr", "ipc:///run/tuer/door_cmd.ipc", "zmq door event publish addr") - flag.StringVar(&doorsub_addr_, "doorsubaddr", "tcp://torwaechter.realraum.at:4242", "zmq door event publish addr") - flag.StringVar(&sensorssub_port_, "sensorsubport", "tcp://*:4243", "zmq public/listen socket addr for incoming sensor data") - flag.StringVar(&pub_port_, "pubport", "tcp://*:4244", "zmq port publishing consodilated events") - flag.StringVar(&keylookup_addr_, "keylookupaddr", "ipc:///run/tuer/door_keyname.ipc", "address to use for key/name lookups") - flag.StringVar(&brain_listen_addr_, "brainlisten", "tcp://*:4245", "address to listen for requests about latest stored event") - flag.BoolVar(&use_syslog_, "syslog", false, "log to syslog local2 facility") - flag.BoolVar(&enable_debuglog_, "debug", false, "enable debug logging") - flag.Parse() + flag.StringVar(&door_cmd_addr_, "doorcmdaddr", "ipc:///run/tuer/door_cmd.ipc", "zmq door event publish addr") + flag.StringVar(&doorsub_addr_, "doorsubaddr", "tcp://torwaechter.realraum.at:4242", "zmq door event publish addr") + flag.StringVar(&sensorssub_port_, "sensorsubport", "tcp://*:4243", "zmq public/listen socket addr for incoming sensor data") + flag.StringVar(&pub_port_, "pubport", "tcp://*:4244", "zmq port publishing consodilated events") + flag.StringVar(&keylookup_addr_, "keylookupaddr", "ipc:///run/tuer/door_keyname.ipc", "address to use for key/name lookups") + flag.StringVar(&brain_listen_addr_, "brainlisten", "tcp://*:4245", "address to listen for requests about latest stored event") + flag.BoolVar(&use_syslog_, "syslog", false, "log to syslog local2 facility") + flag.BoolVar(&enable_debuglog_, "debug", false, "enable debug logging") + flag.Parse() } func main() { - if enable_debuglog_ { LogEnableDebuglog() } - if use_syslog_ { - LogEnableSyslog() - Syslog_.Print("started") - defer Syslog_.Print("exiting") - } + if enable_debuglog_ { + LogEnableDebuglog() + } + if use_syslog_ { + LogEnableSyslog() + Syslog_.Print("started") + defer Syslog_.Print("exiting") + } - zmqctx, sub_in_chans, pub_out_socket, keylookup_socket := ZmqsInit(doorsub_addr_, sensorssub_port_, pub_port_, keylookup_addr_) - if sub_in_chans != nil {defer sub_in_chans.Close()} - defer zmqctx.Close() - if pub_out_socket != nil {defer pub_out_socket.Close()} - if keylookup_socket != nil {defer keylookup_socket.Close()} - if sub_in_chans == nil || pub_out_socket == nil || keylookup_socket == nil { - panic("zmq sockets must not be nil !!") - } + zmqctx, sub_in_chans, pub_out_socket, keylookup_socket := ZmqsInit(doorsub_addr_, sensorssub_port_, pub_port_, keylookup_addr_) + if sub_in_chans != nil { + defer sub_in_chans.Close() + } + defer zmqctx.Close() + if pub_out_socket != nil { + defer pub_out_socket.Close() + } + if keylookup_socket != nil { + defer keylookup_socket.Close() + } + if sub_in_chans == nil || pub_out_socket == nil || keylookup_socket == nil { + panic("zmq sockets must not be nil !!") + } - ps := pubsub.New(10) - defer ps.Shutdown() // ps.Shutdown should be called before zmq_ctx.Close(), since it will cause goroutines to shutdown and close zqm_sockets which is needed for zmq_ctx.Close() to return - //~ ticker := time.NewTicker(time.Duration(5) * time.Minute) + ps := pubsub.New(10) + defer ps.Shutdown() // ps.Shutdown should be called before zmq_ctx.Close(), since it will cause goroutines to shutdown and close zqm_sockets which is needed for zmq_ctx.Close() to return + //~ ticker := time.NewTicker(time.Duration(5) * time.Minute) - store_these_events_chan := ps.Sub("door", "doorcmd", "presence", "sensors", "buttons", "movement") - go BrainCenter(zmqctx, brain_listen_addr_, store_these_events_chan) + store_these_events_chan := ps.Sub("door", "doorcmd", "presence", "sensors", "buttons", "movement") + go BrainCenter(zmqctx, brain_listen_addr_, store_these_events_chan) - go MetaEventRoutine_Movement(ps, 10, 20, 10) - go MetaEventRoutine_Presence(ps, 21, 200) + go MetaEventRoutine_Movement(ps, 10, 20, 10) + go MetaEventRoutine_Presence(ps, 21, 200) - // --- get update on most recent status --- - answ := ZmqsAskQuestionsAndClose(zmqctx, door_cmd_addr_, [][][]byte{[][]byte{[]byte("status")}}) - for _, a := range(answ) { - ParseSocketInputLine(a, ps, keylookup_socket) - } + // --- get update on most recent status --- + answ := ZmqsAskQuestionsAndClose(zmqctx, door_cmd_addr_, [][][]byte{[][]byte{[]byte("status")}}) + for _, a := range answ { + ParseSocketInputLine(a, ps, keylookup_socket) + } - publish_these_events_chan := ps.Sub("door", "doorcmd", "presence", "sensors", "buttons", "movement") - for { - select { - case subin := <- sub_in_chans.In(): - ParseSocketInputLine(subin, ps, keylookup_socket) - //~ case <- ticker.C: - //~ MakeTimeTick(ps) - case event_interface := <- publish_these_events_chan: - data, err := r3events.MarshalEvent2ByteByte(event_interface) - Debug_.Printf("publishing %s",data) - if err != nil { - Syslog_.Print(err) - continue - } - if err := pub_out_socket.Send(data); err != nil { - panic(err) - } - } - } + publish_these_events_chan := ps.Sub("door", "doorcmd", "presence", "sensors", "buttons", "movement") + for { + select { + case subin := <-sub_in_chans.In(): + ParseSocketInputLine(subin, ps, keylookup_socket) + //~ case <- ticker.C: + //~ MakeTimeTick(ps) + case event_interface := <-publish_these_events_chan: + data, err := r3events.MarshalEvent2ByteByte(event_interface) + Debug_.Printf("publishing %s", data) + if err != nil { + Syslog_.Print(err) + continue + } + if err := pub_out_socket.Send(data); err != nil { + panic(err) + } + } + } }