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