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     cmd_chans, sub_chans := ZmqsInit(cmd_port_, sub_port_)
  59     defer cmd_chans.Close()
  60     defer sub_chans.Close()
  61     var listen bool
  62     var ignore_next bool
  63
  64     user_input_chan := make(chan [][]byte, 1)
  65     go LineReader(user_input_chan, os.Stdin)
  66     defer os.Stdin.Close()
  67
  68     for {
  69         select {
  70         case input, input_open := (<- user_input_chan):
  71             if input_open {
  72                 if len(input) == 0 { continue }
  73                  switch string(input[0]) {
  74                     case "listen":
  75                         listen = true
  76                         fmt.Println("Now listening")
  77                     case "quit":
  78                         os.Exit(0)
  79                     default:
  80                         ignore_next = true
  81                         cmd_chans.Out() <- input
  82                         log.Print("input sent")
  83                         reply := <- cmd_chans.In()
  84                         log.Print("reply received")
  85                         fmt.Println(ByteArrayToString(reply))
  86                 }
  87             } else {
  88                 os.Exit(0)
  89             }
  90         case pubsubstuff := <- sub_chans.In():
  91             log.Print("pubsubstuff",pubsubstuff)
  92             if len(pubsubstuff) == 0 { continue}
  93             if ignore_next {
  94                 ignore_next = false
  95                 continue
  96             }
  97             if listen {
  98                 fmt.Println(pubsubstuff)
  99             }
 100         }
 101     }
 102 }