putget/putget.go

   1 //
   2 // Copyright © 2019 Guy M. Allard
   3 //
   4 // Licensed under the Apache License, Version 2.0 (the "License");
   5 // you may not use this file except in compliance with the License.
   6 // You may obtain a copy of the License at
   7 //
   8 //     http://www.apache.org/licenses/LICENSE-2.0
   9 //
  10 // Unless required by applicable law or agreed to in writing, software
  11 // distributed under the License is distributed on an "AS IS" BASIS,
  12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 // See the License for the specific language governing permissions and
  14 // limitations under the License.
  15 //
  16 package main
  17
  18 import (
  19         "log"
  20         "net"
  21         "os"
  22         "strconv"
  23         "time"
  24
  25         "github.com/gmallard/stompngo"
  26         "github.com/gmallard/stompngo/senv"
  27         "github.com/gmallard/stompngo_examples/sngecomm"
  28 )
  29
  30 var (
  31         exampid = "putget: "
  32         ll      = log.New(os.Stdout, "PUGT ", log.Ldate|log.Lmicroseconds|log.Lshortfile)
  33         tag     = "putgemm"
  34         conn    *stompngo.Connection
  35         n       net.Conn
  36         e       error
  37         u2      = false
  38         wns     int64
  39         wd      time.Duration
  40 )
  41
  42 func main() {
  43         st := time.Now()
  44         // Environment variable controls
  45         if os.Getenv("STOMP_2CONN") != "" {
  46                 u2 = true
  47         }
  48         wt := os.Getenv("STOMP_WTIME")
  49         if wt != "" {
  50                 _, err := strconv.ParseInt(wt, 10, 64)
  51                 if err != nil {
  52                         ll.Fatalf("%stag:%s connsess:%s conversion error:%v",
  53                                 exampid, tag, sngecomm.Lcs,
  54                                 e.Error()) // Handle this ......
  55                 } else {
  56                         d, err := time.ParseDuration(wt + "ms")
  57                         if err != nil {
  58                                 ll.Fatalf("%stag:%s connsess:%s parse error:%v",
  59                                         exampid, tag, sngecomm.Lcs,
  60                                         e.Error()) // Handle this ......
  61                         }
  62                         wd = d
  63                         wns = wd.Nanoseconds()
  64                 }
  65         }
  66         // Standard example connect sequence
  67         n, conn, e = sngecomm.CommonConnect(exampid, tag, ll)
  68         if e != nil {
  69                 ll.Fatalf("%stag:%s connsess:%s main_on_connect error:%v",
  70                         exampid, tag, sngecomm.Lcs,
  71                         e.Error()) // Handle this ......
  72         }
  73
  74         ll.Printf("%stag:%s connsess:%s START nmsgs:%d\n",
  75                 exampid, tag, conn.Session(), senv.Nmsgs())
  76         // Put
  77         qname := sngecomm.Dest()
  78         sh := stompngo.Headers{"destination", qname}
  79         amsg := "A message from putget"
  80         // Status message
  81         ll.Printf("%stag:%s connsess:%s will PUT: %s, count: %d\n",
  82                 exampid, tag, conn.Session(), amsg, senv.Nmsgs())
  83         for i := 0; i < senv.Nmsgs(); i++ {
  84                 e = conn.Send(sh, amsg)
  85                 if e != nil {
  86                         ll.Fatalln("Send error:", e)
  87                 }
  88         }
  89         // Possible 2nd connection
  90         if u2 {
  91                 // Standard example disconnect sequence
  92                 e = sngecomm.CommonDisconnect(n, conn, exampid, tag, ll)
  93                 if e != nil {
  94                         ll.Fatalf("%stag:%s connsess:%s main_on_disconnect1 error:%v",
  95                                 exampid, tag, conn.Session(),
  96                                 e.Error()) // Handle this ......
  97                 }
  98                 if wns > 0 {
  99                         time.Sleep(wd)
 100                 }
 101                 //
 102                 n, conn, e = sngecomm.CommonConnect(exampid, tag, ll)
 103                 if e != nil {
 104                         ll.Fatalf("%stag:%s connsess:%s main_on_connect1 error:%v",
 105                                 exampid, tag, sngecomm.Lcs,
 106                                 e.Error()) // Handle this ......
 107                 }
 108         } else if wns > 0 {
 109                 time.Sleep(wd)
 110         }
 111         // Status message
 112         ll.Printf("%stag:%s connsess:%s PUT is done\n",
 113                 exampid, tag, conn.Session())
 114         // Get
 115         id := "putget-subid1"
 116         sc := sngecomm.HandleSubscribe(conn, qname, id, sngecomm.AckMode())
 117         ll.Printf("%stag:%s connsess:%s subscribe_complete id:%v dest:%v\n",
 118                 exampid, tag, conn.Session(),
 119                 id, qname)
 120         //
 121         for i := 0; i < senv.Nmsgs(); i++ {
 122                 md := <-sc
 123                 if md.Error != nil {
 124                         ll.Fatalf("%stag:%s connsess:%s recv_error dest:%s error:%v",
 125                                 exampid, tag, conn.Session(),
 126                                 qname, md.Error) // Handle this ......
 127                 }
 128                 ll.Printf("%stag:%s connsess:%s message is:%s\n",
 129                         exampid, tag, conn.Session(),
 130                         md.Message.BodyString())
 131         }
 132         // Status message
 133         ll.Printf("%stag:%s connsess:%s GET is done\n",
 134                 exampid, tag, conn.Session())
 135         // Standard example disconnect sequence
 136         e = sngecomm.CommonDisconnect(n, conn, exampid, tag, ll)
 137         if e != nil {
 138                 ll.Fatalf("%stag:%s connsess:%s main_on_disconnect error:%v",
 139                         exampid, tag, conn.Session(),
 140                         e.Error()) // Handle this ......
 141         }
 142         ll.Printf("%stag:%s connsess:%s main_elapsed:%v\n",
 143                 exampid, tag, conn.Session(),
 144                 time.Now().Sub(st))
 145 }