receipts/onack/onack.go

   1 //
   2 // Copyright © 2015-2018 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 Show receiving a RECIPT, requested from an ACK.
  19
  20         Examples:
  21
  22                 # Using a broker with all defaults:
  23                 # Host is "localhost"
  24                 # Port is 61613
  25                 # Login is "guest"
  26                 # Passcode is "guest
  27                 # Virtual Host is "localhost"
  28                 # Protocol is 1.1
  29                 go run onack.go
  30
  31                 # Using a broker using a custom host and port:
  32                 STOMP_HOST=tjjackson STOMP_PORT=62613 go run onack.go
  33
  34                 # Using a broker using a custom port and virtual host:
  35                 STOMP_PORT=41613 STOMP_VHOST="/" go run onack.go
  36
  37                 # Using a broker using a custom login and passcode:
  38                 STOMP_LOGIN="userid" STOMP_PASSCODE="t0ps3cr3t" go run onack.go
  39 */
  40 package main
  41
  42 import (
  43         "log"
  44         "os"
  45         "time"
  46         //
  47         "github.com/gmallard/stompngo"
  48         // senv methods could be used in general by stompngo clients.
  49         "github.com/gmallard/stompngo/senv"
  50         // sngecomm methods are used specifically for these example clients.
  51         "github.com/gmallard/stompngo_examples/sngecomm"
  52 )
  53
  54 var (
  55         exampid = "onack: "
  56         ll      = log.New(os.Stdout, "OACK ", log.Ldate|log.Lmicroseconds|log.Lshortfile)
  57
  58         tag = "onackmain"
  59 )
  60
  61 func main() {
  62
  63         // Make sure that the queue used by this example do not exist, or are
  64         // empty.
  65
  66         // Following is a lengthy piece of code.  Read it striaght from top
  67         // to bottom.  There is zero complex logic here.
  68
  69         // What this code will do:
  70         // Phase 1:
  71         // - Connect to a broker
  72         // - Verify a connection spec level
  73         // - Send a single message to the specified queue on that broker
  74         // - Disconnect from that broker
  75         //
  76         // Phase 2:
  77         // - Reconnect to the same broker
  78         // - Subscribe to the specified queue, using "ack:client-individual"
  79         // - Receive a single message
  80         // - Send an ACK, asking for a receipt
  81         //**************************************************************************
  82         // - Receive a RECEIPT // The point of this exercise.
  83         // - Show data from the RECEIPT and verify it // The point of this exercise.
  84         //**************************************************************************
  85         // - Disconnect from the broker
  86
  87         // Start
  88
  89         st := time.Now()
  90
  91         // **************************************** Phase 1
  92         // Set up the connection.
  93         // Standard example connect sequence
  94         n, conn, e := sngecomm.CommonConnect(exampid, tag, ll)
  95         if e != nil {
  96                 ll.Fatalf("%stag:%s connsess:%s main_on_connect error:%v",
  97                         exampid, tag, sngecomm.Lcs,
  98                         e.Error()) // Handle this ......
  99         }
 100
 101         // ****************************************
 102         // App logic here .....
 103
 104         d := sngecomm.Dest()
 105         ll.Printf("%stag:%s connsess:%s destination:%v\n",
 106                 exampid, tag, conn.Session(),
 107                 d)
 108
 109         // ****************************************
 110         // Send exactly one message.
 111         sh := stompngo.Headers{"destination", sngecomm.Dest()}
 112         if senv.Persistent() {
 113                 sh = sh.Add("persistent", "true")
 114         }
 115         m := exampid + " message: "
 116         t := m + "1"
 117         ll.Printf("%stag:%s connsess:%s sending_now body:%v\n",
 118                 exampid, tag, conn.Session(),
 119                 t)
 120         e = conn.Send(sh, t)
 121         if e != nil {
 122                 ll.Fatalf("%stag:%s connsess:%s main_bad_send error:%v",
 123                         exampid, tag, conn.Session(),
 124                         e.Error()) // Handle this ......
 125         }
 126         ll.Printf("%stag:%s connsess:%s send_complete body:%v\n",
 127                 exampid, tag, conn.Session(),
 128                 t)
 129
 130         // ****************************************
 131         // Disconnect from the Stomp server
 132         // Standard example disconnect sequence
 133         e = sngecomm.CommonDisconnect(n, conn, exampid, tag, ll)
 134         if e != nil {
 135                 ll.Fatalf("%stag:%s connsess:%s main_disconnect error:%v",
 136                         exampid, tag, conn.Session(),
 137                         e.Error()) // Handle this ......
 138         }
 139
 140         // **************************************** Phase 2
 141
 142         // Standard example connect sequence
 143         n, conn, e = sngecomm.CommonConnect(exampid, tag, ll)
 144         if e != nil {
 145                 ll.Fatalf("%stag:%s connsess:%s main_on_connect error:%v",
 146                         exampid, tag, sngecomm.Lcs,
 147                         e.Error()) // Handle this ......
 148         }
 149
 150         // ****************************************
 151         // Subscribe here
 152         id := stompngo.Uuid()
 153         // Get the "subscribe channel"
 154         sc := sngecomm.HandleSubscribe(conn, d, id, "client-individual")
 155         ll.Printf("%stag:%s connsess:%s stomp_subscribe_complete\n",
 156                 exampid, tag, conn.Session())
 157
 158         // Get data from the broker
 159         var md stompngo.MessageData // A message data instance
 160         select {
 161         case md = <-sc:
 162         case md = <-conn.MessageData:
 163                 // This would be contain an ERROR or RECEIPT frame.  Both are unexpected
 164                 // in this example.
 165                 ll.Fatalf("%stag:%s connsess:%s bad_frame md:%v",
 166                         exampid, tag, conn.Session(),
 167                         md) // Handle this ......
 168         }
 169         ll.Printf("%stag:%s connsess:%s channel_read_complete\n",
 170                 exampid, tag, conn.Session())
 171
 172         // MessageData has two components:
 173         // a) a Message struct
 174         // b) an Error value.  Check the error value as usual
 175         if md.Error != nil {
 176                 ll.Fatalf("%stag:%s connsess:%s message_error md:%v",
 177                         exampid, tag, conn.Session(),
 178                         md.Error) // Handle this ......
 179         }
 180
 181         ll.Printf("%stag:%s connsess:%s read_message_COMMAND command:%s\n",
 182                 exampid, tag, conn.Session(),
 183                 md.Message.Command)
 184         ll.Printf("%stag:%s connsess:%s read_message_HEADERS headers:%s\n",
 185                 exampid, tag, conn.Session(),
 186                 md.Message.Headers)
 187         ll.Printf("%stag:%s connsess:%s read_message_BODY body:%s\n",
 188                 exampid, tag, conn.Session(),
 189                 string(md.Message.Body))
 190
 191         // Here we need to send an ACK.  Required Headers are different between
 192         // a 1.1 and a 1.2 connection level.
 193         var ah stompngo.Headers
 194         if conn.Protocol() == stompngo.SPL_11 { // 1.1
 195                 ah = ah.Add("subscription", md.Message.Headers.Value("subscription"))
 196                 ah = ah.Add("message-id", md.Message.Headers.Value("message-id"))
 197         } else { // 1.2
 198                 ah = ah.Add("id", md.Message.Headers.Value("ack"))
 199         }
 200         // We are also going to ask for a RECEIPT for the ACK
 201         rid := "receipt-001"
 202         ah = ah.Add("receipt", rid)
 203         //
 204         ll.Printf("%stag:%s connsess:%s ACK_receipt_headers headers:%v\n",
 205                 exampid, tag, conn.Session(),
 206                 ah)
 207         e = conn.Ack(ah)
 208         if e != nil {
 209                 ll.Fatalf("%stag:%s connsess:%s ack_error error:%v",
 210                         exampid, tag, conn.Session(),
 211                         e.Error()) // Handle this ......
 212         }
 213
 214         // ****************************************
 215         // Finally get the RECEIPT.  Where is it?  It is *not* on the "subscribe
 216         // channel".  It is on the connection level MessageData channel.  Why?
 217         // Because the broker does *not* include a "subscription" header in
 218         // RECEIPT frames..
 219         // ****************************************
 220
 221         // ***IMPORTANT***
 222         // ***NOTE*** which channel this RECEIPT MessageData comes in on.
 223
 224         ll.Printf("%stag:%s connsess:%s start_receipt_read\n",
 225                 exampid, tag, conn.Session())
 226         var rd stompngo.MessageData
 227         select {
 228         case rd = <-sc:
 229                 // This would contain a MESSAGE frame.  It is unexpected here
 230                 // in this example.
 231                 ll.Fatalf("%stag:%s connsess:%s bad_frame_channel rd:%v\n",
 232                         exampid, tag, conn.Session(),
 233                         rd) // Handle this ......
 234         case rd = <-conn.MessageData: // RECEIPT frame s/b in the MessageData
 235                 // Step 1 of Verify
 236                 if rd.Message.Command != stompngo.RECEIPT {
 237                         ll.Fatalf("%stag:%s connsess:%s bad_frame_command rd:%v\n",
 238                                 exampid, tag, conn.Session(),
 239                                 rd) // Handle this ......
 240                 }
 241         }
 242         ll.Printf("%stag:%s connsess:%s end_receipt_read\n",
 243                 exampid, tag, conn.Session())
 244
 245         // ****************************************
 246         // Show details about the RECEIPT MessageData struct
 247         ll.Printf("%stag:%s connsess:%s receipt_COMMAND command:%s\n",
 248                 exampid, tag, conn.Session(),
 249                 rd.Message.Command)
 250         ll.Printf("%stag:%s connsess:%s receipt_HEADERS headers:%v\n",
 251                 exampid, tag, conn.Session(),
 252                 rd.Message.Headers)
 253         ll.Printf("%stag:%s connsess:%s receipt_BODY body:%s\n",
 254                 exampid, tag, conn.Session(),
 255                 string(rd.Message.Body))
 256
 257         // Step 2 of Verify
 258         // Verify that the receipt has the id we asked for
 259         if rd.Message.Headers.Value("receipt-id") != rid {
 260                 ll.Fatalf("%stag:%s connsess:%s bad_receipt_id wanted:%v got:%v\n",
 261                         exampid, tag, conn.Session(),
 262                         rid, rd.Message.Headers.Value("receipt-id")) // Handle this ......
 263         }
 264         ll.Printf("%stag:%s connsess:%s receipt_id_verified rid:%s\n",
 265                 exampid, tag, conn.Session(),
 266                 rid)
 267
 268         // ****************************************
 269         // Disconnect from the Stomp server
 270         // Standard example disconnect sequence
 271         e = sngecomm.CommonDisconnect(n, conn, exampid, tag, ll)
 272         if e != nil {
 273                 ll.Fatalf("%stag:%s connsess:%s main_disconnect error:%v",
 274                         exampid, tag, sngecomm.Lcs,
 275                         e.Error()) // Handle this ......
 276         }
 277
 278         ll.Printf("%stag:%s connsess:%s main_elapsed:%v\n",
 279                 exampid, tag, conn.Session(),
 280                 time.Now().Sub(st))
 281
 282 }