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         e := c.Ack(ah)
 212         if e != nil {
 213                 llu.Fatalf("v1:%v v2:%v v3:%v\n", "ack failed", e, c.Protocol())
 214         }
 215         return
 216 }
 217
 218 func ShowRunParms(exampid string) {
 219         llu.Printf("%sHOST:%v\n", exampid, os.Getenv("STOMP_HOST"))
 220         llu.Printf("%sPORT:%v\n", exampid, os.Getenv("STOMP_PORT"))
 221         llu.Printf("%sPROTOCOL:%v\n", exampid, senv.Protocol())
 222         llu.Printf("%sVHOST:%v\n", exampid, senv.Vhost())
 223         llu.Printf("%sNQS:%v\n", exampid, Nqs())
 224         llu.Printf("%sNMSGS:%v\n", exampid, senv.Nmsgs())
 225         llu.Printf("%sSUBCHANCAP:%v\n", exampid, senv.SubChanCap())
 226         llu.Printf("%sRECVFACT:%vn", exampid, RecvFactor())
 227         llu.Printf("%sSENDFACT:%v\n", exampid, SendFactor())
 228         llu.Printf("%sACKMODE:%v\n", exampid, AckMode())
 229 }
 230
 231 // Return broker identity
 232 func ServerIdent(c *stompngo.Connection) string {
 233         cdh := c.ConnectResponse
 234         sr, ok := cdh.Headers.Contains("server")
 235         if !ok {
 236                 return "N/A"
 237         }
 238         return sr
 239 }
 240
 241 // Common example connect logic
 242 func CommonConnect(exampid, tag string, l *log.Logger) (net.Conn,
 243         *stompngo.Connection,
 244         error) {
 245
 246         l.Printf("%stag:%s consess:%v common_connect_starts\n",
 247                 exampid, tag, Lcs)
 248
 249         // Set up the connection.
 250         h, p := senv.HostAndPort()
 251         hap := net.JoinHostPort(h, p)
 252         n, e := net.Dial("tcp", hap)
 253         if e != nil {
 254                 return nil, nil, e
 255         }
 256
 257         l.Printf("%stag:%s connsess:%s common_connect_host_and_port:%v\n",
 258                 exampid, tag, Lcs,
 259                 hap)
 260
 261         // Create connect headers and connect to stompngo
 262         ch := ConnectHeaders()
 263         l.Printf("%stag:%s connsess:%s common_connect_headers headers:%v\n",
 264                 exampid, tag, Lcs,
 265                 ch)
 266         conn, e := stompngo.Connect(n, ch)
 267         if e != nil {
 268                 return nil, nil, e
 269         }
 270         l.Printf("%stag:%s connsess:%s common_connect_complete host:%s port:%s vhost:%s protocol:%s server:%s\n",
 271                 exampid, tag, conn.Session(),
 272                 h, p, senv.Vhost(), conn.Protocol(), ServerIdent(conn))
 273
 274         // Show connect response
 275         l.Printf("%stag:%s connsess:%s common_connect_response connresp:%v\n",
 276                 exampid, tag, conn.Session(),
 277                 conn.ConnectResponse)
 278
 279         // Show heartbeat data (if heart beats are being used)
 280         if senv.Heartbeats() != "" {
 281                 l.Printf("%stag:%s connsess:%s common_connect_heart_beat_send hbsend:%v\n",
 282                         exampid, tag, conn.Session(),
 283                         conn.SendTickerInterval())
 284                 l.Printf("%stag:%s connsess:%s common_connect_heart_beat_recv hbrecv:%v\n",
 285                         exampid, tag, conn.Session(),
 286                         conn.ReceiveTickerInterval())
 287         }
 288
 289         l.Printf("%stag:%s connsess:%s common_connect_local_addr:%s\n",
 290                 exampid, tag, conn.Session(),
 291                 n.LocalAddr().String())
 292         l.Printf("%stag:%s connsess:%s common_connect_remote_addr:%s\n",
 293                 exampid, tag, conn.Session(),
 294                 n.RemoteAddr().String())
 295
 296         //
 297         return n, conn, nil
 298 }
 299
 300 // Common example disconnect logic
 301 func CommonDisconnect(n net.Conn, conn *stompngo.Connection,
 302         exampid, tag string,
 303         l *log.Logger) error {
 304
 305         // Disconnect from the Stomp server
 306         e := conn.Disconnect(stompngo.Headers{})
 307         if e != nil {
 308                 return e
 309         }
 310         l.Printf("%stag:%s consess:%v common_disconnect_complete local_addr:%s remote_addr:%s\n",
 311                 exampid, tag, conn.Session(),
 312                 n.LocalAddr().String(), n.RemoteAddr().String())
 313
 314         // Close the network connection
 315         e = n.Close()
 316         if e != nil {
 317                 return e
 318         }
 319
 320         // Parting messages
 321         l.Printf("%stag:%s consess:%v common_disconnect_network_close_complete\n",
 322                 exampid, tag, conn.Session())
 323         l.Printf("%stag:%s consess:%v common_disconnect_ends\n",
 324                 exampid, tag, conn.Session())
 325
 326         //
 327         return nil
 328 }
 329
 330 // Common example TLS connect logic
 331 func CommonTLSConnect(exampid, tag string, l *log.Logger,
 332         c *tls.Config) (net.Conn, *stompngo.Connection, error) {
 333
 334         l.Printf("%stag:%s consess:%s common_tls_connect_starts\n",
 335                 exampid, tag, Lcs)
 336
 337         // Set up the connection.
 338         h, p := senv.HostAndPort()
 339         hap := net.JoinHostPort(h, p)
 340         n, e := net.Dial("tcp", hap)
 341         if e != nil {
 342                 return nil, nil, e
 343         }
 344
 345         c.ServerName = h // SNI
 346
 347         nc := tls.Client(n, c) // Returns: *tls.Conn : implements net.Conn
 348         e = nc.Handshake()
 349         if e != nil {
 350                 if e.Error() == "EOF" {
 351                         l.Printf("%stag:%s consess:%s common_tls_handshake_EOF_Is_the_broker_port_TLS_enabled? port:%s\n",
 352                                 exampid, tag, Lcs,
 353                                 p)
 354                 }
 355                 l.Fatalf("%stag:%s consess:%s common_tls_handshake_failed error:%v\n",
 356                         exampid, tag, Lcs,
 357                         e.Error())
 358         }
 359         l.Printf("%stag:%s consess:%s common_tls_handshake_complete\n",
 360                 exampid, tag, Lcs)
 361
 362         l.Printf("%stag:%s connsess:%s common_tls_connect_host_and_port:%v\n",
 363                 exampid, tag, Lcs,
 364                 hap)
 365
 366         // Create connect headers and connect to stompngo
 367         ch := ConnectHeaders()
 368         l.Printf("%stag:%s connsess:%s common_tls_connect_headers headers:%v\n",
 369                 exampid, tag, Lcs,
 370                 ch)
 371         conn, e := stompngo.Connect(nc, ch)
 372         if e != nil {
 373                 return nil, nil, e
 374         }
 375         l.Printf("%stag:%s connsess:%s common_tls_connect_complete host:%s vhost:%s protocol:%s server:%s\n",
 376                 exampid, tag, conn.Session(),
 377                 h, senv.Vhost(), conn.Protocol(), ServerIdent(conn))
 378
 379         // Show connect response
 380         l.Printf("%stag:%s connsess:%s common_tls_connect_response connresp:%v\n",
 381                 exampid, tag, conn.Session(),
 382                 conn.ConnectResponse)
 383
 384         // Show heartbeat data (if heart beats are being used)
 385         if senv.Heartbeats() != "" {
 386                 l.Printf("%stag:%s connsess:%s common_tls_connect_heart_beat_send hbsend:%v\n",
 387                         exampid, tag, conn.Session(),
 388                         conn.SendTickerInterval())
 389                 l.Printf("%stag:%s connsess:%s common_tls_connect_heart_beat_recv hbrecv:%v\n",
 390                         exampid, tag, conn.Session(),
 391                         conn.ReceiveTickerInterval())
 392         }
 393
 394         l.Printf("%stag:%s connsess:%s common_tls_connect_local_addr:%s\n",
 395                 exampid, tag, conn.Session(),
 396                 n.LocalAddr().String())
 397         l.Printf("%stag:%s connsess:%s common_tls_connect_remote_addr:%s\n",
 398                 exampid, tag, conn.Session(),
 399                 n.RemoteAddr().String())
 400
 401         //
 402         return nc, conn, nil
 403 }