go/door_client_zmq/main.go

   1 // (c) Bernhard Tittelbach, 2013
   2
   3 package main
   4
   5 import (
   6     "fmt"
   7     "os"
   8     "flag"
   9     "log"
  10     "bufio"
  11     "bytes"
  12 )
  13
  14
  15 // ---------- Main Code -------------
  16
  17 var (
  18     cmd_port_ string
  19     sub_port_ string
  20 )
  21
  22 func usage() {
  23     fmt.Fprintf(os.Stderr, "Usage: door_client_zmq\n")
  24     flag.PrintDefaults()
  25 }
  26
  27 func init() {
  28     flag.StringVar(&cmd_port_, "cmdport", "tcp://127.0.0.1:3232", "zmq command socket path")
  29     flag.StringVar(&sub_port_, "pubport", "pgm://233.252.1.42:4242", "zmq subscribe/listen socket path")
  30     flag.Usage = usage
  31     flag.Parse()
  32 }
  33
  34 func LineReader(out chan <- [][]byte, stdin * os.File) {
  35     linescanner := bufio.NewScanner(stdin)
  36     linescanner.Split(bufio.ScanLines)
  37     defer close(out)
  38     for linescanner.Scan() {
  39         if err := linescanner.Err(); err != nil {
  40             log.Print(err)
  41             return
  42         }
  43         //text := bytes.Fields(linescanner.Bytes()) //this returns a slice (aka pointer, no array deep-copy here)
  44         text := bytes.Fields([]byte(linescanner.Text())) //this allocates a string and slices it -> no race-condition with overwriting any data
  45         if len(text) == 0 {
  46             continue
  47         }
  48         out <- text
  49     }
  50 }
  51
  52 func ByteArrayToString(bb [][]byte) string {
  53     b := bytes.Join(bb, []byte(" "))
  54     return string(b)
  55 }
  56
  57 func main() {
  58     zmqctx, cmd_chans, sub_chans := ZmqsInit(cmd_port_, sub_port_)
  59     defer cmd_chans.Close()
  60     defer sub_chans.Close()
  61     defer zmqctx.Close()
  62     var listen bool
  63     var ignore_next bool
  64
  65     user_input_chan := make(chan [][]byte, 1)
  66     go LineReader(user_input_chan, os.Stdin)
  67     defer os.Stdin.Close()
  68
  69     for {
  70         select {
  71         case input, input_open := (<- user_input_chan):
  72             if input_open {
  73                 if len(input) == 0 { continue }
  74                  switch string(input[0]) {
  75                     case "listen":
  76                         listen = true
  77                         fmt.Println("Now listening")
  78                     case "quit":
  79                         os.Exit(0)
  80                     default:
  81                         ignore_next = true
  82                         cmd_chans.Out() <- input
  83                         reply := <- cmd_chans.In()
  84                         fmt.Println(ByteArrayToString(reply))
  85                 }
  86             } else {
  87                 os.Exit(0)
  88             }
  89         case pubsubstuff := <- sub_chans.In():
  90             log.Print("pubsubstuff",pubsubstuff)
  91             if len(pubsubstuff) == 0 { continue}
  92             if ignore_next {
  93                 ignore_next = false
  94                 continue
  95             }
  96             if listen {
  97                 fmt.Println(pubsubstuff)
  98             }
  99         }
 100     }
 101 }