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 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.Req)
  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.Connect(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(sub_port) > 0 {
  43         sub_sock, err := ctx.Socket(zmq.Sub)
  44         if err != nil {
  45             panic(err)
  46         }
  47         defer func() { if r:= recover(); r != nil { sub_sock.Close(); panic(r) } }()
  48
  49         sub_sock.Subscribe(subfilter)
  50
  51         if err = sub_sock.Connect(sub_port); err != nil {
  52             panic(err)
  53         }
  54
  55         sub_chans = sub_sock.ChannelsBuffer(10)
  56         go zmqsHandleError(sub_chans)
  57     } else {
  58         sub_chans = nil
  59     }
  60
  61     return
  62 }
  63
  64 func zmqsHandleError(chans *zmq.Channels) {
  65     for error := range(chans.Errors()) {
  66         chans.Close()
  67         panic(error)
  68     }
  69 }