subscribe/subscribe.go

   1 //
   2 // Copyright © 2013-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
  19 Subscribe and receive messages from a STOMP broker.
  20
  21         Examples:
  22
  23                 # Subscribe to a broker with all defaults:
  24                 # Host is "localhost"
  25                 # Port is 61613
  26                 # Login is "guest"
  27                 # Passcode is "guest
  28                 # Virtual Host is "localhost"
  29                 # Protocol is 1.2
  30                 go run subscribe.go
  31
  32                 # Subscribe to a broker using STOMP protocol level 1.1:
  33                 STOMP_PROTOCOL=1.1 go run subscribe.go
  34
  35                 # Subscribe to a broker using a custom host and port:
  36                 STOMP_HOST=tjjackson STOMP_PORT=62613 go run subscribe.go
  37
  38                 # Subscribe to a broker using a custom port and virtual host:
  39                 STOMP_PORT=41613 STOMP_VHOST="/" go run subscribe.go
  40
  41                 # Subscribe to a broker using a custom login and passcode:
  42                 STOMP_LOGIN="userid" STOMP_PASSCODE="t0ps3cr3t" go run subscribe.go
  43
  44 */
  45 package main
  46
  47 import (
  48         "log"
  49         "os"
  50         "time"
  51         //
  52         "github.com/gmallard/stompngo"
  53         // senv methods could be used in general by stompngo clients.
  54         "github.com/gmallard/stompngo/senv"
  55         // sngecomm methods are used specifically for these example clients.
  56         "github.com/gmallard/stompngo_examples/sngecomm"
  57 )
  58
  59 var (
  60         exampid = "subscribe: "
  61         ll      = log.New(os.Stdout, "ESUBS ", log.Ldate|log.Lmicroseconds|log.Lshortfile)
  62
  63         tag = "subscribemain"
  64 )
  65
  66 // Connect to a STOMP broker, subscribe and receive some messages and disconnect.
  67 func main() {
  68
  69         st := time.Now()
  70
  71         // Standard example connect sequence
  72         n, conn, e := sngecomm.CommonConnect(exampid, tag, ll)
  73         if e != nil {
  74                 ll.Fatalf("%stag:%s connsess:%s main_on_connect error:%v",
  75                         exampid, tag, sngecomm.Lcs,
  76                         e.Error()) // Handle this ......
  77         }
  78
  79         pbc := sngecomm.Pbc() // Print byte count
  80
  81         // *NOTE* your application functionaltiy goes here!
  82         // With Stomp, you must SUBSCRIBE to a destination in order to receive.
  83         // Subscribe returns a channel of MessageData struct.
  84         // Here we use a common utility routine to handle the differing subscribe
  85         // requirements of each protocol level.
  86         d := senv.Dest()
  87         id := stompngo.Uuid()
  88         sc := sngecomm.HandleSubscribe(conn, d, id, "auto")
  89         ll.Printf("%stag:%s connsess:%s stomp_subscribe_complete\n",
  90                 exampid, tag, conn.Session())
  91         // Read data from the returned channel
  92         var md stompngo.MessageData
  93         for i := 1; i <= senv.Nmsgs(); i++ {
  94
  95                 select {
  96                 case md = <-sc:
  97                 case md = <-conn.MessageData:
  98                         // Frames RECEIPT or ERROR not expected here
  99                         ll.Fatalf("%stag:%s connsess:%s bad_frame error:%v",
 100                                 exampid, tag, conn.Session(),
 101                                 e.Error()) // Handle this ......
 102                 }
 103
 104                 ll.Printf("%stag:%s connsess:%s channel_read_complete\n",
 105                         exampid, tag, conn.Session())
 106                 ll.Printf("%stag:%s connsess:%s message_number:%v\n",
 107                         exampid, tag, conn.Session(),
 108                         i)
 109
 110                 // MessageData has two components:
 111                 // a) a Message struct
 112                 // b) an Error value.  Check the error value as usual
 113                 if md.Error != nil {
 114                         ll.Fatalf("%stag:%s connsess:%s error_read error:%v",
 115                                 exampid, tag, conn.Session(),
 116                                 e.Error()) // Handle this ......
 117                 }
 118                 //
 119                 ll.Printf("%stag:%s connsess:%s frame_type cmd:%s\n",
 120                         exampid, tag, conn.Session(),
 121                         md.Message.Command)
 122
 123                 if md.Message.Command != stompngo.MESSAGE {
 124                         ll.Fatalf("%stag:%s connsess:%s error_frame_type error:%v",
 125                                 exampid, tag, conn.Session(),
 126                                 e.Error()) // Handle this ......
 127                 }
 128                 wh := md.Message.Headers
 129                 for j := 0; j < len(wh)-1; j += 2 {
 130                         ll.Printf("%stag:%s connsess:%s Header:%s:%s\n",
 131                                 exampid, tag, conn.Session(),
 132                                 wh[j], wh[j+1])
 133                 }
 134                 if pbc > 0 {
 135                         maxlen := pbc
 136                         if len(md.Message.Body) < maxlen {
 137                                 maxlen = len(md.Message.Body)
 138                         }
 139                         ss := string(md.Message.Body[0:maxlen])
 140                         ll.Printf("%stag:%s connsess:%s payload body:%s\n",
 141                                 exampid, tag, conn.Session(),
 142                                 ss)
 143                 }
 144                 mbs := string(md.Message.Body)
 145                 if sngecomm.UseEOF() && mbs == sngecomm.EOFMsg {
 146                         ll.Printf("%stag:%s connsess:%s received EOF\n",
 147                                 exampid, tag, conn.Session())
 148                         break
 149                 }
 150         }
 151         // It is polite to unsubscribe, although unnecessary if a disconnect follows.
 152         // Again we use a utility routine to handle the different protocol level
 153         // requirements.
 154         sngecomm.HandleUnsubscribe(conn, d, id)
 155         ll.Printf("%stag:%s connsess:%s stomp_unsubscribe_complete\n",
 156                 exampid, tag, conn.Session())
 157
 158         // Standard example disconnect sequence
 159         e = sngecomm.CommonDisconnect(n, conn, exampid, tag, ll)
 160         if e != nil {
 161                 ll.Fatalf("%stag:%s connsess:%s main_on_disconnect error:%v",
 162                         exampid, tag, conn.Session(),
 163                         e.Error()) // Handle this ......
 164         }
 165
 166         ll.Printf("%stag:%s connsess:%s main_elapsed:%v\n",
 167                 exampid, tag, conn.Session(),
 168                 time.Now().Sub(st))
 169
 170 }