go/door_daemon_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, pub_port string)  (ctx *zmq.Context, cmd_chans, pub_chans *zmq.Channels) {
  13
  14     var err error
  15     ctx, err = zmq.NewContext()
  16     if err != nil {
  17         panic(err)
  18     }
  19     //close only on later panic, otherwise leave open:
  20     defer func(){ if r:= recover(); r != nil { ctx.Close(); panic(r) } }()
  21
  22     if len(cmd_port) > 0 {
  23         cmd_sock, err := ctx.Socket(zmq.Rep)
  24         if err != nil {
  25             panic(err)
  26         }
  27         defer func() { if r:= recover(); r != nil { cmd_sock.Close(); panic(r) } }()
  28
  29         cmd_sock.SetRecvTimeout(2 * time.Second)
  30         cmd_sock.SetSendTimeout(2 * time.Second)
  31
  32             if err = cmd_sock.Bind(cmd_port); err != nil {
  33             panic(err)
  34         }
  35
  36         cmd_chans = cmd_sock.ChannelsBuffer(10)
  37         go zmqsHandleError(cmd_chans)
  38     } else {
  39         cmd_chans = nil
  40     }
  41
  42     if len(pub_port) > 0 {
  43         pub_sock, err := ctx.Socket(zmq.Pub)
  44         if err != nil {
  45             panic(err)
  46         }
  47         defer func() { if r:= recover(); r != nil { pub_sock.Close(); panic(r) } }()
  48
  49         if err = pub_sock.Bind(pub_port); err != nil {
  50             panic(err)
  51         }
  52
  53         pub_chans = pub_sock.ChannelsBuffer(10)
  54         go zmqsHandleError(pub_chans)
  55     } else {
  56         pub_chans = nil
  57     }
  58
  59     return
  60 }
  61
  62 func zmqsHandleError(chans *zmq.Channels) {
  63     for error := range(chans.Errors()) {
  64         chans.Close()
  65         panic(error)
  66     }
  67 }