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