adhoc/reader/reader.go

   1 //
   2 // Copyright © 2011-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 package main
  18
  19 import (
  20         //"fmt"
  21         "log"
  22         "net"
  23         "os"
  24         "time"
  25         //
  26         "github.com/gmallard/stompngo"
  27         // senv methods could be used in general by stompngo clients.
  28         //"github.com/gmallard/stompngo/senv"
  29         // sngecomm methods are used specifically for these example clients.
  30         "github.com/gmallard/stompngo_examples/sngecomm"
  31 )
  32
  33 var (
  34         ll = log.New(os.Stdout, "4EVRD ", log.Ldate|log.Lmicroseconds|log.Lshortfile)
  35
  36         exampid = "reader: "
  37
  38         //
  39         n    net.Conn             // Network Connection
  40         conn *stompngo.Connection // Stomp Connection
  41
  42         lhl = 44
  43
  44         tag = "reader"
  45 )
  46
  47 // A forever reader.
  48 func main() {
  49
  50         st := time.Now()
  51
  52         sngecomm.ShowRunParms(exampid)
  53
  54         // Standard example connect sequence
  55         var e error
  56         n, conn, e = sngecomm.CommonConnect(exampid, tag, ll)
  57         if e != nil {
  58                 if conn != nil {
  59                         ll.Printf("%stag:%s  connsess:%s Connect Response headers:%v body%s\n",
  60                                 exampid, tag, conn.Session(), conn.ConnectResponse.Headers,
  61                                 string(conn.ConnectResponse.Body))
  62                 }
  63                 ll.Fatalf("%stag:%s connsess:%s main_on_connect error:%v",
  64                         exampid, tag, sngecomm.Lcs,
  65                         e.Error()) // Handle this ......
  66         }
  67
  68         //**
  69         qn := 1
  70         ltag := tag + "-receiver"
  71
  72         id := stompngo.Uuid() // A unique subscription ID
  73         d := sngecomm.Dest()
  74
  75         ll.Printf("%stag:%s connsess:%s queue_info id:%v d:%v qnum:%v\n",
  76                 exampid, ltag, conn.Session(),
  77                 id, d, qn)
  78         // Subscribe
  79         sc := sngecomm.HandleSubscribe(conn, d, id, sngecomm.AckMode())
  80         ll.Printf("%stag:%s connsess:%s subscribe_complete id:%v d:%v qnum:%v\n",
  81                 exampid, ltag, conn.Session(),
  82                 id, d, qn)
  83         //
  84         var md stompngo.MessageData
  85         // Receive loop
  86         mc := 1
  87         for {
  88                 ll.Println("========================== ", "Expecting Message:", mc,
  89                         "==========================")
  90                 select {
  91                 case md = <-sc:
  92                 case md = <-conn.MessageData:
  93                         // A RECEIPT or ERROR frame is unexpected here
  94                         ll.Fatalf("%stag:%s connsess:%s bad_frame qnum:%v headers:%v body:%s",
  95                                 exampid, tag, conn.Session(),
  96                                 qn, md.Message.Headers, md.Message.Body) // Handle this ......
  97                 }
  98                 if md.Error != nil {
  99                         ll.Fatalf("%stag:%s connsess:%s recv_error qnum:%v error:%v",
 100                                 exampid, tag, conn.Session(),
 101                                 qn, md.Error) // Handle this ......
 102                 }
 103
 104                 mbs := string(md.Message.Body)
 105                 ll.Printf("%stag:%s connsess:%s next_frame mc:%v command:%v headers:%v body:%v\n",
 106                         exampid, tag, conn.Session(),
 107                         mc, md.Message.Command, md.Message.Headers, mbs) // Handle this ......
 108
 109                 mc++
 110                 if sngecomm.UseEOF() && mbs == sngecomm.EOF_MSG {
 111                         ll.Printf("%stag:%s connsess:%s received EOF\n",
 112                                 exampid, tag, conn.Session())
 113                         break
 114                 }
 115         }
 116
 117         //**
 118
 119         // Standard example disconnect sequence
 120         e = sngecomm.CommonDisconnect(n, conn, exampid, tag, ll)
 121         if e != nil {
 122                 ll.Fatalf("%stag:%s connsess:%s main_on_disconnect error:%v",
 123                         exampid, tag, conn.Session(),
 124                         e.Error()) // Handle this ......
 125         }
 126
 127         sngecomm.ShowStats(exampid, tag, conn)
 128
 129         ll.Printf("%stag:%s connsess:%s main_elapsed:%v\n",
 130                 exampid, tag, conn.Session(),
 131                 time.Now().Sub(st))
 132
 133 }