From 39b62b90f21c12aac821b80b40ee162390ccbbc1 Mon Sep 17 00:00:00 2001 From: Bernhard Tittelbach Date: Fri, 27 Sep 2013 02:27:29 +0000 Subject: [PATCH] ... --- go/r3-eventbroker_zmq/main.go | 24 +++++++++++++++++ go/r3-eventbroker_zmq/sockettoevent.go | 45 +++++++++++++++++++++++--------- go/r3-eventbroker_zmq/zeromq.go | 25 +++++------------- 3 files changed, 63 insertions(+), 31 deletions(-) diff --git a/go/r3-eventbroker_zmq/main.go b/go/r3-eventbroker_zmq/main.go index dfdaff5..3145c6c 100644 --- a/go/r3-eventbroker_zmq/main.go +++ b/go/r3-eventbroker_zmq/main.go @@ -9,6 +9,7 @@ import ( "time" "log/syslog" "log" + pubsub "github.com/tuxychandru/pubsub" ) //~ func StringArrayToByteArray(ss []string) [][]byte { @@ -59,6 +60,29 @@ func main() { Syslog_.Print("started") defer Syslog_.Print("exiting") } + + ps := pubsub.New(3) + ticker := time.NewTicker(time.Duration(5) * time.Minute) + 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 := FormatEventForSocket(event_interface) + if err != nil { + Syslog_.Print(err) + continue + } + if err := pub_out_socket.Send(data); err != nil { + panic(err) + } + } + } //~ nick, err := keylookup_socket.LookupCardIdNick(keyhexid) diff --git a/go/r3-eventbroker_zmq/sockettoevent.go b/go/r3-eventbroker_zmq/sockettoevent.go index e4bfc8e..f2e7a20 100644 --- a/go/r3-eventbroker_zmq/sockettoevent.go +++ b/go/r3-eventbroker_zmq/sockettoevent.go @@ -3,22 +3,22 @@ package main import ( - pubsub "github.com/tuxychandru/pubsub" "regexp" "strconv" - "bufio" "time" //~ "./brain" - "net" + "encoding/json" + pubsub "github.com/tuxychandru/pubsub" + zmq "github.com/vaughan0/go-zmq" ) var ( re_presence_ *regexp.Regexp = regexp.MustCompile("Presence: (yes|no)(?:, (opened|closed), (.+))?") re_state_ *regexp.Regexp = regexp.MustCompile("State: (closed|opened|manual movement|error|reset|timeout after open|timeout after close|opening|closing).*") re_status_ *regexp.Regexp = regexp.MustCompile("Status: (closed|opened) (closed|opened|manual movement|error|reset|timeout after open|timeout after close|opening|closing) (ajar|shut).*") - re_infocard_ *regexp.Regexp = regexp.MustCompile("Info\(card\): card\(([a-fA-F0-9]+)\) (found|not found).*") - re_cardid_ *regexp.Regexp = regexp.MustCompile("card\(([a-fA-F0-9]+)\)") - re_infoajar_ *regexp.Regexp = regexp.MustCompile("Info\(ajar\): door is now (ajar|shut)") + re_infocard_ *regexp.Regexp = regexp.MustCompile("Info\\(card\\): card\\(([a-fA-F0-9]+)\\) (found|not found).*") + re_cardid_ *regexp.Regexp = regexp.MustCompile("card\\(([a-fA-F0-9]+)\\)") + re_infoajar_ *regexp.Regexp = regexp.MustCompile("Info\\(ajar\\): door is now (ajar|shut)") re_command_ *regexp.Regexp = regexp.MustCompile("(open|close|toggle|reset)(?: +(Card|Phone|SSH|ssh))?(?: +(.+))?") re_button_ *regexp.Regexp = regexp.MustCompile("PanicButton|button\\d?") re_temp_ *regexp.Regexp = regexp.MustCompile("temp0: (\\d+\\.\\d+)") @@ -30,6 +30,8 @@ type PresenceUpdate struct { Present bool Ts int64 } +func (s PresenceUpdate) Serialize() string + type DoorLockUpdate struct { DoorID byte @@ -76,7 +78,7 @@ type MovementSensorUpdate struct { Ts int64 } -func parseSocketInputLine_State(lines [][]byte, ps *pubsub.PubSub, ts uint64) { +func parseSocketInputLine_State(lines [][]byte, ps *pubsub.PubSub, ts int64) { switch string(lines[0]) { case "closed": ps.Pub(DoorLockUpdate{0, true, ts}, "door") @@ -94,7 +96,7 @@ func parseSocketInputLine_State(lines [][]byte, ps *pubsub.PubSub, ts uint64) { } -func ParseSocketInputLine(lines [][]byte, ps *pubsub.PubSub) { //, brn *brain.Brain) { +func ParseSocketInputLine(lines [][]byte, ps *pubsub.PubSub, keylookup_socket *zmq.Socket) { //, brn *brain.Brain) { var tidbit interface{} ts := time.Now().Unix() if len(lines) < 1 { return } @@ -104,15 +106,15 @@ func ParseSocketInputLine(lines [][]byte, ps *pubsub.PubSub) { //, brn *brain.Br parseSocketInputLine_State(lines[1:], ps, ts) case "Status:": if len(lines) < 3 { continue } - tidbit = DoorLockUpdate{0, lines[1] == []byte("closed"), ts} + tidbit = DoorLockUpdate{0, string(lines[1]) == "closed", ts} //~ brn.Oboite("door", tidbit) ps.Pub(tidbit, "door") - tidbit = DoorAjarUpdate{0, lines[-1] == []byte("shut"), ts} + tidbit = DoorAjarUpdate{0, string(lines[len(lines)-2]) == "shut", ts} //~ brn.Oboite("door", tidbit) ps.Pub(tidbit, "door") case "Info(card):": if len(lines) < 3 { continue } - if lines[2] != []byte("found") { + if string(lines[2]) != "found" { continue } match_cardid := re_cardid_.FindSubmatch(lines[1]) @@ -128,13 +130,31 @@ func ParseSocketInputLine(lines [][]byte, ps *pubsub.PubSub) { //, brn *brain.Br } case "Info(ajar):": if len(lines) < 5 { continue } - DoorAjarUpdate{0, match_status[4] == []byte("shut"), ts} + tidbit = DoorAjarUpdate{0, string(lines[4]) == "shut", ts} //~ brn.Oboite("door", tidbit) ps.Pub(tidbit, "door") case "open", "close", "toggle", "reset": ps.Pub(DoorCommandEvent{string(lines[0]), string(lines[1]), string(lines[2]), ts},"doorcmd") + case "photo0": + newphoto, err := strconv.ParseInt(string(lines[1]), 10, 32) + if err == nil { + // brn.Oboite("photo0", newphoto) + ps.Pub(IlluminationSensorUpdate{0, newphoto, ts}, "sensors") + } } +} +func MakeTimeTick(ps *pubsub.PubSub) { + ps.Pub(TimeTick{time.Now().Unix()},"time") +} + +func FormatEventForSocket(event_interface interface{}) (data [][]byte, err error) { + msg, err := json.Marshal(data) + if err != nil { + return + } + return [][]byte{msg}, nil +} //~ match_presence := re_presence_.FindStringSubmatch(line) //~ match_status := re_status_.FindStringSubmatch(line) @@ -161,4 +181,3 @@ func ParseSocketInputLine(lines [][]byte, ps *pubsub.PubSub) { //, brn *brain.Br //~ // brn.Oboite("movement", ts) //~ ps.Pub(MovementSensorUpdate{0, ts}, "movements") //~ } -} diff --git a/go/r3-eventbroker_zmq/zeromq.go b/go/r3-eventbroker_zmq/zeromq.go index 2e28d50..55aa201 100644 --- a/go/r3-eventbroker_zmq/zeromq.go +++ b/go/r3-eventbroker_zmq/zeromq.go @@ -5,15 +5,12 @@ package main import ( zmq "github.com/vaughan0/go-zmq" "bytes" - "error" + "errors" ) // ---------- ZeroMQ Code ------------- -type ReqSocket *zmq.Socket -type PubSocket *zmq.Socket - -func ZmqsInit(sub_connect_port, sub_listen_port, pub_port, keylookup_port string) (ctx *zmq.Context, sub_chans *zmq.Channels, pub_sock PubSocket, keylookup_sock ReqSocket) { +func ZmqsInit(sub_connect_port, sub_listen_port, pub_port, keylookup_port string) (ctx *zmq.Context, sub_chans *zmq.Channels, pub_sock *zmq.Socket, keylookup_sock *zmq.Socket) { var err error ctx, err = zmq.NewContext() if err != nil { @@ -22,7 +19,7 @@ func ZmqsInit(sub_connect_port, sub_listen_port, pub_port, keylookup_port string //close only on later panic, otherwise leave open: defer func(){ if r:= recover(); r != nil { ctx.Close(); panic(r) } }() - if len(sub_port) > 0 { + if len(sub_connect_port) > 0 && len(sub_listen_port) > 0 { sub_sock, err := ctx.Socket(zmq.Sub) if err != nil { panic(err) @@ -81,9 +78,8 @@ func zmqsHandleError(chans *zmq.Channels) { } } -func (s ReqSocket) ZmqsRequestAnswer(request [][]byte) (answer []][]byte) { - sock := s.(*zmq.Socket) - if err = sock.Send(request); err != nil { +func (sock *zmq.Socket) ZmqsRequestAnswer(request [][]byte) (answer [][]byte) { + if err := sock.Send(request); err != nil { panic(err) } parts, err := sock.Recv() @@ -93,20 +89,13 @@ func (s ReqSocket) ZmqsRequestAnswer(request [][]byte) (answer []][]byte) { return parts } -func (s PubSocket) ZmqsPublish(msg [][]byte) { - sock := s.(*zmq.Socket) - if err = sock.Send(msg); err != nil { - panic(err) - } -} - -func (s ReqSocket) LookupCardIdNick(hexbytes []byte) (nick string, error) { +func (s *zmq.Socket) LookupCardIdNick(hexbytes []byte) (nick string, error) { answ := s.ZmqsRequestAnswer([][]byte{hexbytes}) if len(answ) == 0 { return "", errors.New("Empty reply received") } if answ[0] == []byte("ERROR") { - return "", errors.New(string(bytes.Join(answ[1:]))) + return "", errors.New(string(bytes.Join(answ[1:],[]byte(" ")))) } if answ[0] != []byte("RESULT") || len(answ) != 3{ return "", errors.New("Unknown reply received") -- 1.7.10.4