receipts/onack/onack.go

   1 //
   2 // Copyright © 2015-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 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         "net"
  45         "os"
  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
  59 func main() {
  60
  61         // Make sure that the queue used by this example do not exist, or are
  62         // empty.
  63
  64         // Following is a lengthy piece of code.  Read it striaght from top
  65         // to bottom.  There is zero comlex logic here.
  66
  67         // Here is what we will do:
  68         // Phase 1:
  69         // - Connect to a broker
  70         // - Verify a connection spec level
  71         // - Send a single message to the specified queue on that broker
  72         // - Disconnect from that broker
  73         //
  74         // Phase 2:
  75         // - Reconnect to the same broker
  76         // - Subscribe to the specified queue, using "ack:client-individual"
  77         // - Receive a single message
  78         // - Send an ACK, asking for a receipt
  79         // - Receive a RECEIPT # The point of this exercise.
  80         // - Show data from the RECEIPT and verify it
  81         // - Disconnect from the broker
  82
  83         ll.Printf("%s starts\n", exampid)
  84
  85         // **************************************** Phase 1
  86         // Set up the connection.
  87         h, p := senv.HostAndPort()
  88         hap := net.JoinHostPort(h, p)
  89         n, e := net.Dial("tcp", hap)
  90         if e != nil {
  91                 ll.Fatalf("%s %s\n", exampid, e.Error()) // Handle this ......
  92         }
  93         ll.Printf("%s dial1_complete hap:%s\n",
  94                 exampid, hap)
  95         ch := sngecomm.ConnectHeaders()
  96         conn, e := stompngo.Connect(n, ch)
  97         if e != nil {
  98                 ll.Fatalf("%s %s\n", exampid, e.Error()) // Handle this ......
  99         }
 100
 101         if conn.Protocol() == stompngo.SPL_10 {
 102                 ll.Fatalf("%s v1:%v\n", exampid, "STOMP 1.0 not supported for this example")
 103         }
 104         ll.Printf("%s connsess:%s stomp_connect1_complete protocol:%s\n",
 105                 exampid, conn.Session(),
 106                 conn.Protocol())
 107
 108         // ****************************************
 109         // App logic here .....
 110
 111         d := senv.Dest()
 112         // Prep
 113         ll.Printf("%s connsess:%s d%s\n",
 114                 exampid, conn.Session(),
 115                 d)
 116
 117         // ****************************************
 118         // Send exactly one message.
 119         sh := stompngo.Headers{"destination", senv.Dest()}
 120         if senv.Persistent() {
 121                 sh = sh.Add("persistent", "true")
 122         }
 123         m := exampid + " message: "
 124         t := m + "1"
 125
 126         ll.Printf("%s connsess:%s sending_now t:%s\n",
 127                 exampid, conn.Session(),
 128                 t)
 129         e = conn.Send(sh, t)
 130         if e != nil {
 131                 ll.Fatalf("%s v1:%v v2:%v\n", exampid, "bad send", e) // Handle this ...
 132         }
 133
 134         ll.Printf("%s connsess:%s send_complete t:%s\n",
 135                 exampid, conn.Session(),
 136                 t)
 137
 138         // ****************************************
 139         // Disconnect from the Stomp server
 140         e = conn.Disconnect(stompngo.Headers{})
 141         if e != nil {
 142                 ll.Fatalf("%s %s\n", exampid, e.Error()) // Handle this ......
 143         }
 144         ll.Printf("%s connsess:%s stomp_disconnect1_complete t:%s\n",
 145                 exampid, conn.Session(),
 146                 t)
 147         // Close the network connection
 148         e = n.Close()
 149         if e != nil {
 150                 ll.Fatalf("%s %s\n", exampid, e.Error()) // Handle this ......
 151         }
 152         ll.Printf("%s connsess:%s net_close1_complete t:%s\n",
 153                 exampid, conn.Session(),
 154                 t)
 155
 156         // **************************************** Phase 2
 157
 158         n, e = net.Dial("tcp", net.JoinHostPort(h, p))
 159         if e != nil {
 160                 ll.Fatalf("%s %s\n", exampid, e.Error()) // Handle this ......
 161         }
 162
 163         ll.Printf("%s dial2_complete hap:%s\n",
 164                 exampid, net.JoinHostPort(h, p))
 165
 166         conn, e = stompngo.Connect(n, ch)
 167         if e != nil {
 168                 ll.Fatalf("%s  f4v:%v\n", exampid, 10, e) // Handle this ......
 169         }
 170         ll.Printf("%s connsess:%s stomp_connect2_complete protocol:%s\n",
 171                 exampid, conn.Session(),
 172                 conn.Protocol())
 173
 174         // ****************************************
 175         // Subscribe here
 176         id := stompngo.Uuid()
 177
 178         var md stompngo.MessageData // A message data instance
 179
 180         // Get the "subscribe channel"
 181         sc := sngecomm.HandleSubscribe(conn, d, id, "client-individual")
 182         ll.Printf("%s connsess:%s stomp_subscribe_complete\n",
 183                 exampid, conn.Session())
 184
 185         // Get data from the broker
 186         select {
 187         case md = <-sc:
 188         case md = <-conn.MessageData:
 189                 // This would be contain an ERROR or RECEIPT frame.  Both are unexpected
 190                 // in this example.
 191                 ll.Fatalf("%s v1:%v\n", exampid, md) // Handle this
 192         }
 193         ll.Printf("%s connsess:%s channel_read_complete\n",
 194                 exampid, conn.Session())
 195
 196         // MessageData has two components:
 197         // a) a Message struct
 198         // b) an Error value.  Check the error value as usual
 199         if md.Error != nil {
 200                 ll.Fatalf("%s v1:%v\n", exampid, md.Error) // Handle this
 201         }
 202
 203         ll.Printf("%s connsess:%s read_message_COMMAND command:%s\n",
 204                 exampid, conn.Session(),
 205                 md.Message.Command)
 206         ll.Printf("%s connsess:%s read_message_HEADERS headers:%v\n",
 207                 exampid, conn.Session(),
 208                 md.Message.Headers)
 209         ll.Printf("%s connsess:%s read_message_BODY body:%s\n",
 210                 exampid, conn.Session(),
 211                 string(md.Message.Body))
 212
 213         // Here we need to send an ACK.  Required Headers are different between
 214         // a 1.1 and a 1.2 connection level.
 215         var ah stompngo.Headers
 216         if conn.Protocol() == stompngo.SPL_11 { // 1.1
 217                 ah = ah.Add("subscription", md.Message.Headers.Value("subscription"))
 218                 ah = ah.Add("message-id", md.Message.Headers.Value("message-id"))
 219         } else { // 1.2
 220                 ah = ah.Add("id", md.Message.Headers.Value("ack"))
 221         }
 222
 223         // We are also going to ask for a RECEIPT for the ACK
 224         rid := "1"
 225         ah = ah.Add("receipt", rid)
 226         //
 227         e = conn.Ack(ah)
 228         if e != nil {
 229                 ll.Fatalf("%s %s\n", exampid, e.Error()) // Handle this ......
 230         }
 231
 232         // ****************************************
 233         // Finally get the RECEIPT.  Where is it?  It is *not* on the "subscribe
 234         // channel".  It is on the connection level MessageData channel.  Why?
 235         // Because the broker does *not* include a "subscription" header in
 236         // RECEIPT frames..
 237         // ****************************************
 238
 239         // ***IMPORTANT***
 240         // ***NOTE*** which channel this RECEIPT MessageData comes in on.
 241         var rd stompngo.MessageData
 242
 243         ll.Printf("%s connsess:%s start_receipt_read\n",
 244                 exampid, conn.Session())
 245         select {
 246         case rd = <-sc:
 247                 // This would contain a MESSAGE frame.  It is unexpected here
 248                 // in this example.
 249                 ll.Fatalf("%s v1:%v\n", exampid, md) // Handle this
 250         case rd = <-conn.MessageData: // RECEIPT frame s/b in the MessageData
 251                 if rd.Message.Command != stompngo.RECEIPT {
 252                         ll.Fatalf("%s v1:%v\n", exampid, md) // Handle this
 253                 }
 254         }
 255         ll.Printf("%s connsess:%s end_receipt_read\n",
 256                 exampid, conn.Session())
 257         // ****************************************
 258         // Show stuff about the RECEIPT MessageData struct
 259
 260         ll.Printf("%s connsess:%s receipt_COMMAND command:%s\n",
 261                 exampid, conn.Session(),
 262                 rd.Message.Command)
 263         ll.Printf("%s connsess:%s receipt_HEADERS headers:%v\n",
 264                 exampid, conn.Session(),
 265                 rd.Message.Headers)
 266         ll.Printf("%s connsess:%s receipt_BODY body:%s\n",
 267                 exampid, conn.Session(),
 268                 string(rd.Message.Body))
 269
 270         // ****************************************
 271         // Disconnect from the Stomp server
 272         e = conn.Disconnect(stompngo.Headers{})
 273         if e != nil {
 274                 ll.Fatalf("%s %s\n", exampid, e.Error()) // Handle this ......
 275         }
 276
 277         ll.Printf("%s connsess:%s stomp_disconnect2_complete\n",
 278                 exampid, conn.Session())
 279         // Close the network connection
 280         e = n.Close()
 281         if e != nil {
 282                 ll.Fatalf("%s %s\n", exampid, e.Error()) // Handle this ......
 283         }
 284
 285         ll.Printf("%s connsess:%s net_close2_complete\n",
 286                 exampid, conn.Session())
 287 }