sngecomm/utilities.go

   1 //
   2 // Copyright © 2016 Guy M. Alluard
   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 Package sngecomm provides common functionality used in the stompngo_examples
  19 project.
  20 */
  21 package sngecomm
  22
  23 import (
  24         "crypto/rand"
  25         "crypto/tls"
  26         "log"
  27         "math/big"
  28         "net"
  29         "os"
  30         "strings"
  31         //
  32         "github.com/gmallard/stompngo"
  33         "github.com/gmallard/stompngo/senv"
  34 )
  35
  36 var (
  37         llu = log.New(os.Stdout, "UTIL ", log.Ldate|log.Lmicroseconds|log.Lshortfile)
  38         Lcs = "NotAvailable"
  39 )
  40
  41 // Provide connect headers
  42 func ConnectHeaders() stompngo.Headers {
  43         h := stompngo.Headers{}
  44         l := senv.Login()
  45         if l != "" {
  46                 h = h.Add("login", l)
  47         }
  48         pc := senv.Passcode()
  49         if pc != "" {
  50                 h = h.Add("passcode", pc)
  51         }
  52         //
  53         p := senv.Protocol()
  54         if p != stompngo.SPL_10 { // 1.1 and 1.2
  55                 h = h.Add("accept-version", p).Add("host", senv.Vhost())
  56                 hb := senv.Heartbeats()
  57                 if hb != "" {
  58                         h = h.Add("heart-beat", hb)
  59                 }
  60         }
  61         //
  62
  63         return h
  64 }
  65
  66 // Show connection metrics.
  67 func ShowStats(exampid, tag string, conn *stompngo.Connection) {
  68         r := conn.FramesRead()
  69         br := conn.BytesRead()
  70         w := conn.FramesWritten()
  71         bw := conn.BytesWritten()
  72         s := conn.Running().Seconds()
  73         n := conn.Running().Nanoseconds()
  74         llu.Printf("%stag:%s frame_read_count:%v\n", exampid, tag, r)
  75         llu.Printf("%stag:%s bytes_read:%v\n", exampid, tag, br)
  76         llu.Printf("%stag:%s frame_write_count:%v\n", exampid, tag, w)
  77         llu.Printf("%stag:%s bytes_written:%v\n", exampid, tag, bw)
  78         llu.Printf("%stag:%s current_duration(ns):%v\n", exampid, tag, n)
  79
  80         llu.Printf("%stag:%s current_duration(sec):%20.6f\n", exampid, tag, s)
  81         llu.Printf("%stag:%s frame_reads/sec:%20.6f\n", exampid, tag, float64(r)/s)
  82         llu.Printf("%stag:%s bytes_read/sec:%20.6f\n", exampid, tag, float64(br)/s)
  83         llu.Printf("%stag:%s frame_writes/sec:%20.6f\n", exampid, tag, float64(w)/s)
  84         llu.Printf("%stag:%s bytes_written/sec:%20.6f\n", exampid, tag, float64(bw)/s)
  85 }
  86
  87 // Get a value between min amd max
  88 func ValueBetween(min, max int64, fact float64) int64 {
  89         rt, _ := rand.Int(rand.Reader, big.NewInt(max-min)) // Ignore errors here
  90         return int64(fact * float64(min+rt.Int64()))
  91 }
  92
  93 // Dump a TLS Configuration Struct
  94 func DumpTLSConfig(exampid string, c *tls.Config, n *tls.Conn) {
  95         llu.Printf("%s TLSConfig:\n", exampid)
  96         llu.Printf("%s Rand:%#v\n", exampid, c.Rand)
  97         llu.Printf("%s Time:%v\n", exampid, c.Time)
  98         llu.Printf("%s Certificates:%#v\n", exampid, c.Certificates)
  99         llu.Printf("%s NameToCertificate:%#v\n", exampid, c.NameToCertificate)
 100         llu.Printf("%s RootCAs:%#v\n", exampid, c.RootCAs)
 101         llu.Printf("%s NextProtos:%v\n", exampid, c.NextProtos)
 102         llu.Printf("%s ServerName:%v\n", exampid, c.ServerName)
 103         llu.Printf("%s ClientAuth:%v\n", exampid, c.ClientAuth)
 104         llu.Printf("%s ClientCAs:%v#\n", exampid, c.ClientCAs)
 105         llu.Printf("%s CipherSuites:%#v\n", exampid, c.CipherSuites)
 106         llu.Printf("%s PreferServerCipherSuites:%v\n", exampid, c.PreferServerCipherSuites)
 107         llu.Printf("%s SessionTicketsDisabled:%v\n", exampid, c.SessionTicketsDisabled)
 108         llu.Printf("%s SessionTicketKey:%#v\n", exampid, c.SessionTicketKey)
 109
 110         // Idea Embelluished From:
 111         // https://groups.google.com/forum/#!topic/golang-nuts/TMNdOxugbTY
 112         cs := n.ConnectionState()
 113         llu.Printf("%s HandshakeComplete:%v\n", exampid, cs.HandshakeComplete)
 114         llu.Printf("%s DidResume:%v\n", exampid, cs.DidResume)
 115         llu.Printf("%s CipherSuite:%d(0x%X)\n", exampid, cs.CipherSuite, cs.CipherSuite)
 116         llu.Printf("%s NegotiatedProtocol:%v\n", exampid, cs.NegotiatedProtocol)
 117         llu.Printf("%s NegotiatedProtocolIsMutual:%v\n", exampid, cs.NegotiatedProtocolIsMutual)
 118         // llu.Printf("%s ServerName:%v\n", exampid, cs.ServerName) // Server side only
 119         // Portions of any Peer Certificates present
 120         certs := cs.PeerCertificates
 121         if certs == nil || len(certs) < 1 {
 122                 llu.Printf("Could not get server's certificate from the TLS connection.\n")
 123                 return
 124         }
 125         if len(certs) == 1 {
 126                 llu.Printf("%s There is %d Server Cert:\n", exampid, len(certs))
 127         } else {
 128                 llu.Printf("%s There are %d Server Certs:\n", exampid, len(certs))
 129         }
 130
 131         for i, cert := range certs {
 132                 llu.Printf("%s Certificate chain:%d\n", exampid, i)
 133                 llu.Printf("%s Common Name:%s\n", exampid, cert.Subject.CommonName)
 134                 //
 135                 llu.Printf("%s Subject Alternative Names (DNSNames):\n", exampid)
 136                 for idx, dnsn := range cert.DNSNames {
 137                         llu.Printf("%s \tNumber:%d, DNS Name:%s\n", exampid, idx+1, dnsn)
 138                 }
 139                 //
 140                 llu.Printf("%s Subject Alternative Names (Emailaddresses):\n", exampid)
 141                 for idx, enn := range cert.EmailAddresses {
 142                         llu.Printf("%s \tNumber:%d, DNS Name:%s\n", exampid, idx+1, enn)
 143                 }
 144                 //
 145                 llu.Printf("%s Subject Alternative Names (IPAddresses):\n", exampid)
 146                 for idx, ipadn := range cert.IPAddresses {
 147                         llu.Printf("%s \tNumber:%d, DNS Name:%v\n", exampid, idx+1, ipadn)
 148                 }
 149                 //
 150                 llu.Printf("%s Valid Not Before:%s\n", exampid, cert.NotBefore.Local().String())
 151                 llu.Printf("%s Valid Not After:%s\n", exampid, cert.NotAfter.Local().String())
 152                 llu.Println(strings.Repeat("=", 80))
 153         }
 154
 155 }
 156
 157 // Handle a subscribe for the different protocol levels.
 158 func HandleSubscribe(c *stompngo.Connection, d, i, a string) <-chan stompngo.MessageData {
 159         h := stompngo.Headers{"destination", d, "ack", a}
 160         //
 161         switch c.Protocol() {
 162         case stompngo.SPL_12:
 163                 // Add required id header
 164                 h = h.Add("id", i)
 165         case stompngo.SPL_11:
 166                 // Add required id header
 167                 h = h.Add("id", i)
 168         case stompngo.SPL_10:
 169                 // Nothing else to do here
 170         default:
 171                 llu.Fatalf("v1:%v v2:%v\n", "subscribe invalid protocol level, should not happen")
 172         }
 173         //
 174         r, e := c.Subscribe(h)
 175         if e != nil {
 176                 llu.Fatalf("v1:%v v2:%v\n", "subscribe failed", e)
 177         }
 178         return r
 179 }
 180
 181 // Handle a unsubscribe for the different protocol levels.
 182 func HandleUnsubscribe(c *stompngo.Connection, d, i string) {
 183         sbh := stompngo.Headers{}
 184         //
 185         switch c.Protocol() {
 186         case stompngo.SPL_12:
 187                 sbh = sbh.Add("id", i)
 188         case stompngo.SPL_11:
 189                 sbh = sbh.Add("id", i)
 190         case stompngo.SPL_10:
 191                 sbh = sbh.Add("destination", d)
 192         default:
 193                 llu.Fatalf("v1:%v v2:%v\n", "unsubscribe invalid protocol level, should not happen")
 194         }
 195         e := c.Unsubscribe(sbh)
 196         if e != nil {
 197                 llu.Fatalf("v1:%v v2:%v d:%v\n", "unsubscribe failed", e, d)
 198         }
 199         return
 200 }
 201
 202 // Handle ACKs for the different protocol levels.
 203 func HandleAck(c *stompngo.Connection, h stompngo.Headers, id string) {
 204         ah := stompngo.Headers{}
 205         //
 206         switch c.Protocol() {
 207         case stompngo.SPL_12:
 208                 ah = ah.Add("id", h.Value("ack"))
 209         case stompngo.SPL_11:
 210                 ah = ah.Add("message-id", h.Value("message-id")).Add("subscription", id)
 211         case stompngo.SPL_10:
 212                 ah = ah.Add("message-id", h.Value("message-id"))
 213         default:
 214                 llu.Fatalf("v1:%v v2:%v\n", "ack invalid protocol level, should not happen")
 215         }
 216         if cv, ok := h.Contains(stompngo.HK_RECEIPT); ok {
 217                 ah = ah.Add(stompngo.HK_RECEIPT, cv)
 218         }
 219         e := c.Ack(ah)
 220         if e != nil {
 221                 llu.Fatalf("v1:%v v2:%v v3:%v\n", "ack failed", e, c.Protocol())
 222         }
 223         return
 224 }
 225
 226 func ShowRunParms(exampid string) {
 227         llu.Printf("%sHOST:%v\n", exampid, os.Getenv("STOMP_HOST"))
 228         llu.Printf("%sPORT:%v\n", exampid, os.Getenv("STOMP_PORT"))
 229         llu.Printf("%sPROTOCOL:%v\n", exampid, senv.Protocol())
 230         llu.Printf("%sVHOST:%v\n", exampid, senv.Vhost())
 231         llu.Printf("%sNQS:%v\n", exampid, Nqs())
 232         llu.Printf("%sNMSGS:%v\n", exampid, senv.Nmsgs())
 233         llu.Printf("%sSUBCHANCAP:%v\n", exampid, senv.SubChanCap())
 234         llu.Printf("%sRECVFACT:%v\n", exampid, RecvFactor())
 235         llu.Printf("%sSENDFACT:%v\n", exampid, SendFactor())
 236         llu.Printf("%sRECVWAIT:%t\n", exampid, RecvWait())
 237         llu.Printf("%sSENDWAIT:%t\n", exampid, SendWait())
 238         llu.Printf("%sACKMODE:%v\n", exampid, AckMode())
 239 }
 240
 241 // Return broker identity
 242 func ServerIdent(c *stompngo.Connection) string {
 243         cdh := c.ConnectResponse
 244         sr, ok := cdh.Headers.Contains("server")
 245         if !ok {
 246                 return "N/A"
 247         }
 248         return sr
 249 }
 250
 251 // Common example connect logic
 252 func CommonConnect(exampid, tag string, l *log.Logger) (net.Conn,
 253         *stompngo.Connection,
 254         error) {
 255
 256         l.Printf("%stag:%s consess:%v common_connect_starts\n",
 257                 exampid, tag, Lcs)
 258
 259         // Set up the connection.
 260         h, p := senv.HostAndPort()
 261         hap := net.JoinHostPort(h, p)
 262         n, e := net.Dial("tcp", hap)
 263         if e != nil {
 264                 return nil, nil, e
 265         }
 266
 267         l.Printf("%stag:%s connsess:%s common_connect_host_and_port:%v\n",
 268                 exampid, tag, Lcs,
 269                 hap)
 270
 271         // Create connect headers and connect to stompngo
 272         ch := ConnectHeaders()
 273         l.Printf("%stag:%s connsess:%s common_connect_headers headers:%v\n",
 274                 exampid, tag, Lcs,
 275                 ch)
 276         conn, e := stompngo.Connect(n, ch)
 277         if e != nil {
 278                 return nil, conn, e
 279         }
 280         SetLogger(conn) // Maybe set a connection logger
 281         l.Printf("%stag:%s connsess:%s common_connect_complete host:%s port:%s vhost:%s protocol:%s server:%s\n",
 282                 exampid, tag, conn.Session(),
 283                 h, p, senv.Vhost(), conn.Protocol(), ServerIdent(conn))
 284
 285         // Show connect response
 286         l.Printf("%stag:%s connsess:%s common_connect_response connresp:%v\n",
 287                 exampid, tag, conn.Session(),
 288                 conn.ConnectResponse)
 289
 290         // Heartbeat Data
 291         l.Printf("%stag:%s connsess:%s common_connect_heart_beat_send hbsend:%d\n",
 292                 exampid, tag, conn.Session(),
 293                 conn.SendTickerInterval())
 294         l.Printf("%stag:%s connsess:%s common_connect_heart_beat_recv hbrecv:%d\n",
 295                 exampid, tag, conn.Session(),
 296                 conn.ReceiveTickerInterval())
 297
 298         l.Printf("%stag:%s connsess:%s common_connect_local_addr:%s\n",
 299                 exampid, tag, conn.Session(),
 300                 n.LocalAddr().String())
 301         l.Printf("%stag:%s connsess:%s common_connect_remote_addr:%s\n",
 302                 exampid, tag, conn.Session(),
 303                 n.RemoteAddr().String())
 304
 305         //
 306         return n, conn, nil
 307 }
 308
 309 // Common example disconnect logic
 310 func CommonDisconnect(n net.Conn, conn *stompngo.Connection,
 311         exampid, tag string,
 312         l *log.Logger) error {
 313
 314         // Disconnect from the Stomp server
 315         e := conn.Disconnect(stompngo.Headers{})
 316         if e != nil {
 317                 return e
 318         }
 319         l.Printf("%stag:%s consess:%v common_disconnect_complete local_addr:%s remote_addr:%s\n",
 320                 exampid, tag, conn.Session(),
 321                 n.LocalAddr().String(), n.RemoteAddr().String())
 322
 323         // Close the network connection
 324         e = n.Close()
 325         if e != nil {
 326                 return e
 327         }
 328
 329         // Parting messages
 330         l.Printf("%stag:%s consess:%v common_disconnect_network_close_complete\n",
 331                 exampid, tag, conn.Session())
 332         l.Printf("%stag:%s consess:%v common_disconnect_ends\n",
 333                 exampid, tag, conn.Session())
 334
 335         //
 336         return nil
 337 }
 338
 339 // Common example TLS connect logic
 340 func CommonTLSConnect(exampid, tag string, l *log.Logger,
 341         c *tls.Config) (net.Conn, *stompngo.Connection, error) {
 342
 343         l.Printf("%stag:%s consess:%s common_tls_connect_starts\n",
 344                 exampid, tag, Lcs)
 345
 346         // Set up the connection.
 347         h, p := senv.HostAndPort()
 348         hap := net.JoinHostPort(h, p)
 349         n, e := net.Dial("tcp", hap)
 350         if e != nil {
 351                 return nil, nil, e
 352         }
 353
 354         c.ServerName = h // SNI
 355
 356         nc := tls.Client(n, c) // Returns: *tls.Conn : implements net.Conn
 357         e = nc.Handshake()
 358         if e != nil {
 359                 if e.Error() == "EOF" {
 360                         l.Printf("%stag:%s consess:%s common_tls_handshake_EOF_Is_the_broker_port_TLS_enabled? port:%s\n",
 361                                 exampid, tag, Lcs,
 362                                 p)
 363                 }
 364                 l.Fatalf("%stag:%s consess:%s common_tls_handshake_failed error:%v\n",
 365                         exampid, tag, Lcs,
 366                         e.Error())
 367         }
 368         l.Printf("%stag:%s consess:%s common_tls_handshake_complete\n",
 369                 exampid, tag, Lcs)
 370
 371         l.Printf("%stag:%s connsess:%s common_tls_connect_host_and_port:%v\n",
 372                 exampid, tag, Lcs,
 373                 hap)
 374
 375         // Create connect headers and connect to stompngo
 376         ch := ConnectHeaders()
 377         l.Printf("%stag:%s connsess:%s common_tls_connect_headers headers:%v\n",
 378                 exampid, tag, Lcs,
 379                 ch)
 380         conn, e := stompngo.Connect(nc, ch)
 381         if e != nil {
 382                 return nil, nil, e
 383         }
 384         SetLogger(conn)
 385         l.Printf("%stag:%s connsess:%s common_tls_connect_complete host:%s vhost:%s protocol:%s server:%s\n",
 386                 exampid, tag, conn.Session(),
 387                 h, senv.Vhost(), conn.Protocol(), ServerIdent(conn))
 388
 389         // Show connect response
 390         l.Printf("%stag:%s connsess:%s common_tls_connect_response connresp:%v\n",
 391                 exampid, tag, conn.Session(),
 392                 conn.ConnectResponse)
 393
 394         // Show heartbeat data (if heart beats are being used)
 395         if senv.Heartbeats() != "" {
 396                 l.Printf("%stag:%s connsess:%s common_tls_connect_heart_beat_send hbsend:%v\n",
 397                         exampid, tag, conn.Session(),
 398                         conn.SendTickerInterval())
 399                 l.Printf("%stag:%s connsess:%s common_tls_connect_heart_beat_recv hbrecv:%v\n",
 400                         exampid, tag, conn.Session(),
 401                         conn.ReceiveTickerInterval())
 402         }
 403
 404         l.Printf("%stag:%s connsess:%s common_tls_connect_local_addr:%s\n",
 405                 exampid, tag, conn.Session(),
 406                 n.LocalAddr().String())
 407         l.Printf("%stag:%s connsess:%s common_tls_connect_remote_addr:%s\n",
 408                 exampid, tag, conn.Session(),
 409                 n.RemoteAddr().String())
 410
 411         //
 412         return nc, conn, nil
 413 }
 414
 415 // Example destination
 416 func Dest() string {
 417         d := senv.Dest()
 418         if os.Getenv("STOMP_ARTEMIS") == "" {
 419                 return d
 420         }
 421         pref := "jms.queue"
 422         if strings.Index(d, "topic") >= 0 {
 423                 pref = "jms.topic"
 424         }
 425         return pref + strings.Replace(d, "/", ".", -1)
 426 }
 427
 428 // Set Logger
 429 func SetLogger(conn *stompngo.Connection) {
 430         if Logger() != "" {
 431                 ul := log.New(os.Stdout, Logger()+" ", log.Ldate|log.Lmicroseconds|log.Lshortfile)
 432                 conn.SetLogger(ul)
 433         }
 434 }