go/door_client_zmq/zeromq.go

   1 // (c) Bernhard Tittelbach, 2013
   2
   3 package main
   4
   5 import (
   6     zmq "github.com/vaughan0/go-zmq"
   7     "time"
   8  )
   9
  10 // ---------- ZeroMQ Code -------------
  11
  12 func ZmqsInit(cmd_port, sub_port string)  (ctx *zmq.Context, cmd_chans, sub_chans *zmq.Channels) {
  13     var subfilter []byte
  14     var err error
  15     ctx, err = zmq.NewContext()
  16     if err != nil {
  17         panic(err)
  18     }
  19     //close only on panic, otherwise leave open:
  20     defer func(){ if r:= recover(); r != nil { ctx.Close(); panic(r) } }()
  21
  22     cmd_sock, err := ctx.Socket(zmq.Req)
  23     if err != nil {
  24         panic(err)
  25     }
  26     defer func() { if r:= recover(); r != nil { cmd_sock.Close(); panic(r) } }()
  27
  28     cmd_sock.SetRecvTimeout(2 * time.Second)
  29     cmd_sock.SetSendTimeout(2 * time.Second)
  30
  31     sub_sock, err := ctx.Socket(zmq.Sub)
  32     if err != nil {
  33         panic(err)
  34     }
  35     defer func() { if r:= recover(); r != nil { sub_sock.Close(); panic(r) } }()
  36
  37     if err = cmd_sock.Connect(cmd_port); err != nil {
  38         panic(err)
  39     }
  40
  41     sub_sock.Subscribe(subfilter)
  42
  43     if err = sub_sock.Connect(sub_port); err != nil {
  44         panic(err)
  45     }
  46
  47     cmd_chans = cmd_sock.ChannelsBuffer(10)
  48     sub_chans = cmd_sock.ChannelsBuffer(10)
  49
  50     go zmqsHandleError(cmd_chans, sub_chans)
  51     return
  52 }
  53
  54 func zmqsHandleError(cmd_chans, sub_chans *zmq.Channels) {
  55     for {
  56         select {
  57             case cmd_error := <- cmd_chans.Errors():
  58                 cmd_chans.Close()
  59                 sub_chans.Close()
  60                 panic(cmd_error)
  61             case sub_error := <- sub_chans.Errors():
  62                 cmd_chans.Close()
  63                 sub_chans.Close()
  64                 panic(sub_error)
  65         }
  66     }
  67 }