adhoc/varmGetter/noPackMod/noPMod1/noPMod1.go

   1 //
   2 // Copyright © 2016 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
  17 /*
  18
  19 Function: read the code :-).
  20
  21 */
  22 package main
  23
  24 import (
  25         "fmt"
  26         "log"
  27         "os"
  28         "time"
  29         //
  30         "github.com/gmallard/stompngo"
  31         "github.com/gmallard/stompngo/senv"
  32         "github.com/gmallard/stompngo_examples/sngecomm"
  33 )
  34
  35 var (
  36         exampid = "varmGetter: "
  37         ll      = log.New(os.Stdout, "VRMGSC ", log.Ldate|log.Lmicroseconds)
  38         tag     = "vrmgmain"
  39         unsub   = true
  40         dodisc  = true
  41         ar      = false // Want ACK RECEIPT
  42         session = ""
  43         qcb     []chan bool
  44         wlp     = "publish: message: "
  45 )
  46
  47 func init() {
  48         if os.Getenv("VMG_NOUNSUB") != "" {
  49                 unsub = false
  50         }
  51         if os.Getenv("VMG_NODISC") != "" {
  52                 dodisc = false
  53         }
  54         if os.Getenv("VMG_GETAR") != "" {
  55                 ar = true
  56         }
  57 }
  58
  59 // Connect to a STOMP broker, subscribe and receive some messages and disconnect.
  60 func main() {
  61
  62         st := time.Now()
  63
  64         // Standard example connect sequence
  65         n, conn, e := sngecomm.CommonConnect(exampid, tag, ll)
  66         if e != nil {
  67                 ll.Fatalf("%stag:%s connsess:%s main_on_connect error:%v",
  68                         exampid, tag, sngecomm.Lcs,
  69                         e.Error()) // Handle this ......
  70         }
  71         session = conn.Session()
  72         //******************
  73         nqs := sngecomm.Nqs()
  74         for qn := 1; qn <= nqs; qn++ {
  75                 runNextQueue(qn, conn)
  76         }
  77         //******************
  78
  79         // Standard example disconnect sequence
  80         if dodisc {
  81                 e = sngecomm.CommonDisconnect(n, conn, exampid, tag, ll)
  82                 if e != nil {
  83                         ll.Fatalf("%stag:%s connsess:%s main_on_disconnect error:%v",
  84                                 exampid, tag, session,
  85                                 e.Error()) // Handle this ......
  86                 }
  87                 ll.Printf("%stag:%s connsess:%s disconnect_receipt:%v\n",
  88                         exampid, tag, session,
  89                         conn.DisconnectReceipt)
  90         } else {
  91                 ll.Printf("%stag:%s connsess:%s skipping_disconnect\n",
  92                         exampid, tag, session)
  93         }
  94
  95         ll.Printf("%stag:%s connsess:%s main_elapsed:%v\n",
  96                 exampid, tag, session,
  97                 time.Now().Sub(st))
  98 }
  99
 100 func runNextQueue(qn int, conn *stompngo.Connection) {
 101
 102         qns := fmt.Sprintf("%d", qn) // string number of the queue
 103         conn.SetLogger(ll)           // stompngo logging
 104         pbc := sngecomm.Pbc()        // Print byte count
 105         d := senv.Dest() + qns       // Destination
 106         id := stompngo.Uuid()        // A unique name/id
 107         nmsgs := qn                  // int number of messages to get, same as queue number
 108         am := sngecomm.AckMode()     // ACK mode to use on SUBSCRIBE
 109         nfa := true                  // Need "final" ACK (possiby reset below)
 110         wh := stompngo.Headers{}     // Starting SUBSCRIBE headers
 111
 112         // Sanity check ACK mode
 113         if conn.Protocol() == stompngo.SPL_10 &&
 114                 am == stompngo.AckModeClientIndividual {
 115                 ll.Fatalf("%stag:%s connsess:%s invalid_ack_mode am:%v proto:%v\n",
 116                         exampid, tag, session,
 117                         am, conn.Protocol()) //
 118         }
 119         // Do not do final ACK if running ACKs are issued
 120         if am == stompngo.AckModeClientIndividual ||
 121                 am == stompngo.AckModeAuto {
 122                 nfa = false
 123         }
 124
 125         // Show run parameters
 126         ll.Printf("%stag:%s connsess:%s run_parms\n\tqns:%v\n\tpbc:%v\n\td:%v\n\tid:%v\n\tnmsgs:%v\n\tam:%v\n\tnfa:%v\n\twh:%v\n",
 127                 exampid, tag, session,
 128                 qns, pbc, d, id, nmsgs, am, nfa, wh)
 129
 130         // Run SUBSCRIBE
 131         sc := doSubscribe(conn, d, id, am, wh)
 132         ll.Printf("%stag:%s connsess:%s stomp_subscribe_complete\n",
 133                 exampid, tag, session)
 134
 135         var md stompngo.MessageData  // Message data from basic read
 136         var lmd stompngo.MessageData // Possible save (copy) of received data
 137         mc := 1                      // Initial message number
 138
 139         // Loop for the requested number of messages
 140 GetLoop:
 141         for {
 142                 ll.Printf("%stag:%s connsess:%s start_of_read_loop mc:%v nmsgs:%v\n",
 143                         exampid, tag, session, mc, nmsgs)
 144
 145                 mcs := fmt.Sprintf("%d", mc) // string number message count
 146
 147                 // Get something from the stompngo read routine
 148                 select {
 149                 case md = <-sc:
 150                 case md = <-conn.MessageData:
 151                         //
 152                         if md.Message.Command == stompngo.RECEIPT {
 153                                 ll.Printf("%stag:%s connsess:%s have_receipt md:%v\n",
 154                                         exampid, tag, session,
 155                                         md)
 156                                 continue GetLoop
 157                         }
 158                         ll.Fatalf("%stag:%s connsess:%s ERROR_frame hdrs:%v body:%v\n",
 159                                 exampid, tag, session,
 160                                 md.Message.Headers, string(md.Message.Body)) // Handle this ......
 161                 }
 162
 163                 // Save message data for possible use in the final ACK
 164                 if mc == nmsgs && nfa {
 165                         lmd = md // Save last message
 166                 }
 167
 168                 // Basic loop logging
 169                 ll.Printf("%stag:%s connsess:%s channel_read_complete qn:%d mc:%d\n",
 170                         exampid, tag, session,
 171                         qn, mc)
 172                 ll.Printf("%stag:%s connsess:%s message_number:%v\n",
 173                         exampid, tag, session,
 174                         mc)
 175
 176                 // Check if reader returned any error
 177                 if md.Error != nil {
 178                         ll.Fatalf("%stag:%s connsess:%s error_read error:%v",
 179                                 exampid, tag, session,
 180                                 md.Error) // Handle this ......
 181                 }
 182
 183                 // Show frame type
 184                 ll.Printf("%stag:%s connsess:%s frame_type cmd:%s\n",
 185                         exampid, tag, session,
 186                         md.Message.Command)
 187
 188                 // Pure sanity check:  this should *never* happen based on logic
 189                 // above.
 190                 if md.Message.Command != stompngo.MESSAGE {
 191                         ll.Fatalf("%stag:%s connsess:%s error_frame_type md:%v",
 192                                 exampid, tag, session,
 193                                 md) // Handle this ......
 194                 }
 195
 196                 // Show Message Headers
 197                 wh := md.Message.Headers
 198                 for j := 0; j < len(wh)-1; j += 2 {
 199                         ll.Printf("%stag:%s connsess:%s Header:%s:%s\n",
 200                                 exampid, tag, session,
 201                                 wh[j], wh[j+1])
 202                 }
 203                 // Show (part of) Message Body
 204                 if pbc > 0 {
 205                         maxlen := pbc
 206                         if len(md.Message.Body) < maxlen {
 207                                 maxlen = len(md.Message.Body)
 208                         }
 209                         ss := string(md.Message.Body[0:maxlen])
 210                         ll.Printf("%stag:%s connsess:%s payload body:%s\n",
 211                                 exampid, tag, session,
 212                                 ss)
 213                 }
 214
 215                 // Sanity check this message payload
 216                 wm := wlp + mcs // The left part plus the (string) meassage number]
 217                 bm := string(md.Message.Body)
 218                 if bm != wm {
 219                         ll.Fatalf("%stag:%s connsess:%s error_message_payload\n\tGot %s\n\tWant%s\n",
 220                                 exampid, tag, session,
 221                                 bm, wm) // Handle this ......
 222                 } else {
 223                         ll.Printf("%stag:%s connsess:%s  matched_body_string\n%s\n%s\n",
 224                                 exampid, tag, session,
 225                                 bm, wm) // Handle this ......)
 226                 }
 227
 228                 // Run individual ACK if required
 229                 if am == stompngo.AckModeClientIndividual {
 230                         wh := md.Message.Headers // Copy Headers
 231                         if ar {                  // ACK receipt wanted
 232                                 wh = wh.Add(stompngo.HK_RECEIPT, "rwanted-"+mcs)
 233                         }
 234                         sngecomm.HandleAck(conn, wh, id)
 235                         ll.Printf("%stag:%s connsess:%s  individual_ack_complete mc:%v headers:%v\n",
 236                                 exampid, tag, session,
 237                                 mc, md.Message.Headers)
 238
 239                 }
 240
 241                 // Check for end of loop condition
 242                 if mc == nmsgs {
 243                         break
 244                 }
 245
 246                 // Increment loop/message counter
 247                 mc++
 248         }
 249
 250         qc := make(chan bool)
 251         go drainSub(session, sc, qc, qn)
 252         dd := false
 253
 254         // Issue the final ACK if needed
 255         if nfa {
 256                 wh := lmd.Message.Headers // Copy Headers
 257                 if ar {                   // ACK receipt wanted
 258                         wh = wh.Add(stompngo.HK_RECEIPT, "rwanted-fin")
 259                 }
 260                 sngecomm.HandleAck(conn, wh, id)
 261                 ll.Printf("%stag:%s connsess:%s  final_ack_complete\n",
 262                         exampid, tag, session)
 263                 if ar {
 264                         ll.Printf("%stag:%s connsess:%s  ack_receive_wait_for_drain qn:%d\n",
 265                                 exampid, tag, session,
 266                                 qn)
 267                         <-qc // Wait for drainSub
 268                         ll.Printf("%stag:%s connsess:%s  ack_receive_drain_complete qn:%d\n",
 269                                 exampid, tag, session,
 270                                 qn)
 271                         dd = true
 272                         getReceipt(conn)
 273                 }
 274         }
 275
 276         if !dd {
 277                 ll.Printf("%stag:%s connsess:%s  message_loop_wait_for_drain qn:%d\n",
 278                         exampid, tag, session,
 279                         qn)
 280                 <-qc
 281                 ll.Printf("%stag:%s connsess:%s  message_loop_drain_complete qn:%d\n",
 282                         exampid, tag, session,
 283                         qn)
 284         }
 285
 286         // Unsubscribe (may be skipped if requested)
 287         if unsub {
 288                 sngecomm.HandleUnsubscribe(conn, d, id)
 289                 ll.Printf("%stag:%s connsess:%s stomp_unsubscribe_complete\n",
 290                         exampid, tag, session)
 291         } else {
 292                 ll.Printf("%stag:%s connsess:%s skipping_unsubscribe\n",
 293                         exampid, tag, session)
 294         }
 295 }
 296
 297 // Handle a subscribe for the different protocol levels.
 298 func doSubscribe(c *stompngo.Connection, d, id, a string, h stompngo.Headers) <-chan stompngo.MessageData {
 299         h = h.Add("destination", d).Add("ack", a)
 300         //
 301         switch c.Protocol() {
 302         case stompngo.SPL_12:
 303                 // Add required id header
 304                 h = h.Add("id", id)
 305         case stompngo.SPL_11:
 306                 // Add required id header
 307                 h = h.Add("id", id)
 308         case stompngo.SPL_10:
 309                 // Nothing else to do here
 310         default:
 311                 ll.Fatalf("v1:%v\n", "subscribe invalid protocol level, should not happen")
 312         }
 313         //
 314         r, e := c.Subscribe(h)
 315         if e != nil {
 316                 ll.Fatalf("subscribe failed err:[%v]\n", e)
 317         }
 318         return r
 319 }
 320
 321 // Get receipt
 322 func getReceipt(conn *stompngo.Connection) {
 323         rd := <-conn.MessageData
 324         ll.Printf("%stag:%s connsess:%s have_receipt_sub md:%v\n",
 325                 exampid, tag, session,
 326                 rd)
 327 }
 328
 329 // Drain a subscription
 330 func drainSub(s string, sc <-chan stompngo.MessageData, dc chan bool, qn int) {
 331         ll.Printf("%stag:%s connsess:%s drain_starts qn:%d\n",
 332                 exampid, tag, s,
 333                 qn)
 334         tmr := time.NewTimer(1 * time.Second) // Time to wait, YMMV
 335         q := false
 336         for {
 337                 select {
 338                 case md := <-sc:
 339                         ll.Printf("%stag:%s connsess:%s drained md:%v\n",
 340                                 exampid, tag, s,
 341                                 md)
 342                 case _ = <-tmr.C:
 343                         q = true
 344                 }
 345                 if q {
 346                         dc <- true
 347                         break
 348                 }
 349         }
 350         ll.Printf("%stag:%s connsess:%s drain_ends qn:%d\n",
 351                 exampid, tag, s,
 352                 qn)
 353 }