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         }
  57         //
  58         hb := senv.Heartbeats()
  59         if hb != "" {
  60                 h = h.Add("heart-beat", hb)
  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         llu.Printf("%s Server Certs:\n", exampid)
 126         for i, cert := range certs {
 127                 llu.Printf("%s Certificate chain:%d\n", exampid, i)
 128                 llu.Printf("%s Common Name:%s\n", exampid, cert.Subject.CommonName)
 129                 //
 130                 llu.Printf("%s Subject Alternative Names (DNSNames):\n", exampid)
 131                 for idx, dnsn := range cert.DNSNames {
 132                         llu.Printf("%s \tNumber:%d, DNS Name:%s\n", exampid, idx+1, dnsn)
 133                 }
 134                 //
 135                 llu.Printf("%s Subject Alternative Names (Emailaddresses):\n", exampid)
 136                 for idx, enn := range cert.EmailAddresses {
 137                         llu.Printf("%s \tNumber:%d, DNS Name:%s\n", exampid, idx+1, enn)
 138                 }
 139                 //
 140                 llu.Printf("%s Subject Alternative Names (IPAddresses):\n", exampid)
 141                 for idx, ipadn := range cert.IPAddresses {
 142                         llu.Printf("%s \tNumber:%d, DNS Name:%v\n", exampid, idx+1, ipadn)
 143                 }
 144                 //
 145                 llu.Printf("%s Valid Not Before:%s\n", exampid, cert.NotBefore.Local().String())
 146                 llu.Printf("%s Valid Not After:%s\n", exampid, cert.NotAfter.Local().String())
 147                 llu.Println(strings.Repeat("=", 80))
 148         }
 149
 150 }
 151
 152 // Handle a subscribe for the different protocol levels.
 153 func HandleSubscribe(c *stompngo.Connection, d, i, a string) <-chan stompngo.MessageData {
 154         h := stompngo.Headers{"destination", d, "ack", a}
 155         //
 156         switch c.Protocol() {
 157         case stompngo.SPL_12:
 158                 // Add required id header
 159                 h = h.Add("id", i)
 160         case stompngo.SPL_11:
 161                 // Add required id header
 162                 h = h.Add("id", i)
 163         case stompngo.SPL_10:
 164                 // Nothing else to do here
 165         default:
 166                 llu.Fatalf("v1:%v v2:%v\n", "subscribe invalid protocol level, should not happen")
 167         }
 168         //
 169         r, e := c.Subscribe(h)
 170         if e != nil {
 171                 llu.Fatalf("v1:%v v2:%v\n", "subscribe failed", e)
 172         }
 173         return r
 174 }
 175
 176 // Handle a unsubscribe for the different protocol levels.
 177 func HandleUnsubscribe(c *stompngo.Connection, d, i string) {
 178         sbh := stompngo.Headers{}
 179         //
 180         switch c.Protocol() {
 181         case stompngo.SPL_12:
 182                 sbh = sbh.Add("id", i)
 183         case stompngo.SPL_11:
 184                 sbh = sbh.Add("id", i)
 185         case stompngo.SPL_10:
 186                 sbh = sbh.Add("destination", d)
 187         default:
 188                 llu.Fatalf("v1:%v v2:%v\n", "unsubscribe invalid protocol level, should not happen")
 189         }
 190         e := c.Unsubscribe(sbh)
 191         if e != nil {
 192                 llu.Fatalf("v1:%v v2:%v d:%v\n", "unsubscribe failed", e, d)
 193         }
 194         return
 195 }
 196
 197 // Handle ACKs for the different protocol levels.
 198 func HandleAck(c *stompngo.Connection, h stompngo.Headers, id string) {
 199         ah := stompngo.Headers{}
 200         //
 201         switch c.Protocol() {
 202         case stompngo.SPL_12:
 203                 ah = ah.Add("id", h.Value("ack"))
 204         case stompngo.SPL_11:
 205                 ah = ah.Add("message-id", h.Value("message-id")).Add("subscription", id)
 206         case stompngo.SPL_10:
 207                 ah = ah.Add("message-id", h.Value("message-id"))
 208         default:
 209                 llu.Fatalf("v1:%v v2:%v\n", "ack invalid protocol level, should not happen")
 210         }
 211         if cv, ok := h.Contains(stompngo.HK_RECEIPT); ok {
 212                 ah = ah.Add(stompngo.HK_RECEIPT, cv)
 213         }
 214         e := c.Ack(ah)
 215         if e != nil {
 216                 llu.Fatalf("v1:%v v2:%v v3:%v\n", "ack failed", e, c.Protocol())
 217         }
 218         return
 219 }
 220
 221 func ShowRunParms(exampid string) {
 222         llu.Printf("%sHOST:%v\n", exampid, os.Getenv("STOMP_HOST"))
 223         llu.Printf("%sPORT:%v\n", exampid, os.Getenv("STOMP_PORT"))
 224         llu.Printf("%sPROTOCOL:%v\n", exampid, senv.Protocol())
 225         llu.Printf("%sVHOST:%v\n", exampid, senv.Vhost())
 226         llu.Printf("%sNQS:%v\n", exampid, Nqs())
 227         llu.Printf("%sNMSGS:%v\n", exampid, senv.Nmsgs())
 228         llu.Printf("%sSUBCHANCAP:%v\n", exampid, senv.SubChanCap())
 229         llu.Printf("%sRECVFACT:%vn", exampid, RecvFactor())
 230         llu.Printf("%sSENDFACT:%v\n", exampid, SendFactor())
 231         llu.Printf("%sRECVWAIT:%t\n", exampid, RecvWait())
 232         llu.Printf("%sSENDWAIT:%t\n", exampid, SendWait())
 233         llu.Printf("%sACKMODE:%v\n", exampid, AckMode())
 234 }
 235
 236 // Return broker identity
 237 func ServerIdent(c *stompngo.Connection) string {
 238         cdh := c.ConnectResponse
 239         sr, ok := cdh.Headers.Contains("server")
 240         if !ok {
 241                 return "N/A"
 242         }
 243         return sr
 244 }
 245
 246 // Common example connect logic
 247 func CommonConnect(exampid, tag string, l *log.Logger) (net.Conn,
 248         *stompngo.Connection,
 249         error) {
 250
 251         l.Printf("%stag:%s consess:%v common_connect_starts\n",
 252                 exampid, tag, Lcs)
 253
 254         // Set up the connection.
 255         h, p := senv.HostAndPort()
 256         hap := net.JoinHostPort(h, p)
 257         n, e := net.Dial("tcp", hap)
 258         if e != nil {
 259                 return nil, nil, e
 260         }
 261
 262         l.Printf("%stag:%s connsess:%s common_connect_host_and_port:%v\n",
 263                 exampid, tag, Lcs,
 264                 hap)
 265
 266         // Create connect headers and connect to stompngo
 267         ch := ConnectHeaders()
 268         l.Printf("%stag:%s connsess:%s common_connect_headers headers:%v\n",
 269                 exampid, tag, Lcs,
 270                 ch)
 271         conn, e := stompngo.Connect(n, ch)
 272         if e != nil {
 273                 return nil, nil, e
 274         }
 275         l.Printf("%stag:%s connsess:%s common_connect_complete host:%s port:%s vhost:%s protocol:%s server:%s\n",
 276                 exampid, tag, conn.Session(),
 277                 h, p, senv.Vhost(), conn.Protocol(), ServerIdent(conn))
 278
 279         // Show connect response
 280         l.Printf("%stag:%s connsess:%s common_connect_response connresp:%v\n",
 281                 exampid, tag, conn.Session(),
 282                 conn.ConnectResponse)
 283
 284         // Show heartbeat data (if heart beats are being used)
 285         if senv.Heartbeats() != "" {
 286                 l.Printf("%stag:%s connsess:%s common_connect_heart_beat_send hbsend:%v\n",
 287                         exampid, tag, conn.Session(),
 288                         conn.SendTickerInterval())
 289                 l.Printf("%stag:%s connsess:%s common_connect_heart_beat_recv hbrecv:%v\n",
 290                         exampid, tag, conn.Session(),
 291                         conn.ReceiveTickerInterval())
 292         }
 293
 294         l.Printf("%stag:%s connsess:%s common_connect_local_addr:%s\n",
 295                 exampid, tag, conn.Session(),
 296                 n.LocalAddr().String())
 297         l.Printf("%stag:%s connsess:%s common_connect_remote_addr:%s\n",
 298                 exampid, tag, conn.Session(),
 299                 n.RemoteAddr().String())
 300
 301         //
 302         return n, conn, nil
 303 }
 304
 305 // Common example disconnect logic
 306 func CommonDisconnect(n net.Conn, conn *stompngo.Connection,
 307         exampid, tag string,
 308         l *log.Logger) error {
 309
 310         // Disconnect from the Stomp server
 311         e := conn.Disconnect(stompngo.Headers{})
 312         if e != nil {
 313                 return e
 314         }
 315         l.Printf("%stag:%s consess:%v common_disconnect_complete local_addr:%s remote_addr:%s\n",
 316                 exampid, tag, conn.Session(),
 317                 n.LocalAddr().String(), n.RemoteAddr().String())
 318
 319         // Close the network connection
 320         e = n.Close()
 321         if e != nil {
 322                 return e
 323         }
 324
 325         // Parting messages
 326         l.Printf("%stag:%s consess:%v common_disconnect_network_close_complete\n",
 327                 exampid, tag, conn.Session())
 328         l.Printf("%stag:%s consess:%v common_disconnect_ends\n",
 329                 exampid, tag, conn.Session())
 330
 331         //
 332         return nil
 333 }
 334
 335 // Common example TLS connect logic
 336 func CommonTLSConnect(exampid, tag string, l *log.Logger,
 337         c *tls.Config) (net.Conn, *stompngo.Connection, error) {
 338
 339         l.Printf("%stag:%s consess:%s common_tls_connect_starts\n",
 340                 exampid, tag, Lcs)
 341
 342         // Set up the connection.
 343         h, p := senv.HostAndPort()
 344         hap := net.JoinHostPort(h, p)
 345         n, e := net.Dial("tcp", hap)
 346         if e != nil {
 347                 return nil, nil, e
 348         }
 349
 350         c.ServerName = h // SNI
 351
 352         nc := tls.Client(n, c) // Returns: *tls.Conn : implements net.Conn
 353         e = nc.Handshake()
 354         if e != nil {
 355                 if e.Error() == "EOF" {
 356                         l.Printf("%stag:%s consess:%s common_tls_handshake_EOF_Is_the_broker_port_TLS_enabled? port:%s\n",
 357                                 exampid, tag, Lcs,
 358                                 p)
 359                 }
 360                 l.Fatalf("%stag:%s consess:%s common_tls_handshake_failed error:%v\n",
 361                         exampid, tag, Lcs,
 362                         e.Error())
 363         }
 364         l.Printf("%stag:%s consess:%s common_tls_handshake_complete\n",
 365                 exampid, tag, Lcs)
 366
 367         l.Printf("%stag:%s connsess:%s common_tls_connect_host_and_port:%v\n",
 368                 exampid, tag, Lcs,
 369                 hap)
 370
 371         // Create connect headers and connect to stompngo
 372         ch := ConnectHeaders()
 373         l.Printf("%stag:%s connsess:%s common_tls_connect_headers headers:%v\n",
 374                 exampid, tag, Lcs,
 375                 ch)
 376         conn, e := stompngo.Connect(nc, ch)
 377         if e != nil {
 378                 return nil, nil, e
 379         }
 380         l.Printf("%stag:%s connsess:%s common_tls_connect_complete host:%s vhost:%s protocol:%s server:%s\n",
 381                 exampid, tag, conn.Session(),
 382                 h, senv.Vhost(), conn.Protocol(), ServerIdent(conn))
 383
 384         // Show connect response
 385         l.Printf("%stag:%s connsess:%s common_tls_connect_response connresp:%v\n",
 386                 exampid, tag, conn.Session(),
 387                 conn.ConnectResponse)
 388
 389         // Show heartbeat data (if heart beats are being used)
 390         if senv.Heartbeats() != "" {
 391                 l.Printf("%stag:%s connsess:%s common_tls_connect_heart_beat_send hbsend:%v\n",
 392                         exampid, tag, conn.Session(),
 393                         conn.SendTickerInterval())
 394                 l.Printf("%stag:%s connsess:%s common_tls_connect_heart_beat_recv hbrecv:%v\n",
 395                         exampid, tag, conn.Session(),
 396                         conn.ReceiveTickerInterval())
 397         }
 398
 399         l.Printf("%stag:%s connsess:%s common_tls_connect_local_addr:%s\n",
 400                 exampid, tag, conn.Session(),
 401                 n.LocalAddr().String())
 402         l.Printf("%stag:%s connsess:%s common_tls_connect_remote_addr:%s\n",
 403                 exampid, tag, conn.Session(),
 404                 n.RemoteAddr().String())
 405
 406         //
 407         return nc, conn, nil
 408 }