connect.go

   1 //
   2 // Copyright © 2011-2019 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 stompngo
  18
  19 import (
  20         "bufio"
  21         "log"
  22         "os"
  23
  24         // "fmt"
  25         "net"
  26         "time"
  27
  28         "github.com/gmallard/stompngo/senv"
  29 )
  30
  31 /*
  32         Wrapper for primary STOMP Connect function that returns an interface.
  33 */
  34 func NewConnector(n net.Conn, h Headers) (STOMPConnector, error) {
  35         return Connect(n, h)
  36 }
  37
  38 /*
  39         Primary STOMP Connect.
  40
  41         For STOMP 1.1+ the Headers parameter MUST contain the headers required
  42         by the specification.  Those headers are not magically inferred.
  43
  44         Example:
  45                 // Obtain a network connection
  46                 n, e := net.Dial(NetProtoTCP, "localhost:61613")
  47                 if e != nil {
  48                         // Do something sane ...
  49                 }
  50                 h := stompngo.Headers{} // A STOMP 1.0 connection request
  51                 c, e := stompngo.Connect(n, h)
  52                 if e != nil {
  53                         // Do something sane ...
  54                 }
  55                 // Use c
  56
  57         Example:
  58                 // Obtain a network connection
  59                 n, e := net.Dial(NetProtoTCP, "localhost:61613")
  60                 if e != nil {
  61                         // Do something sane ...
  62                 }
  63                 h := stompngo.Headers{HK_ACCEPT_VERSION, "1.1",
  64                         HK_HOST, "localhost"} // A STOMP 1.1 connection
  65                 c, e := stompngo.Connect(n, h)
  66                 if e != nil {
  67                         // Do something sane ...
  68                 }
  69                 // Use c
  70 */
  71 func Connect(n net.Conn, h Headers) (*Connection, error) {
  72         if h == nil {
  73                 return nil, EHDRNIL
  74         }
  75         if e := h.Validate(); e != nil {
  76                 return nil, e
  77         }
  78         if _, ok := h.Contains(HK_RECEIPT); ok {
  79                 return nil, ENORECPT
  80         }
  81         ch := h.Clone()
  82         //fmt.Printf("CONDB01\n")
  83         c := &Connection{netconn: n,
  84                 input:             make(chan MessageData, 1),
  85                 output:            make(chan wiredata),
  86                 connected:         false,
  87                 session:           "",
  88                 protocol:          SPL_10,
  89                 subs:              make(map[string]*subscription),
  90                 DisconnectReceipt: MessageData{},
  91                 ssdc:              make(chan struct{}),
  92                 wtrsdc:            make(chan struct{}),
  93                 scc:               1,
  94                 dld:               &deadlineData{}}
  95
  96         // Basic metric data
  97         c.mets = &metrics{st: time.Now()}
  98
  99         // Assumed for now
 100         c.MessageData = c.input
 101
 102         // Check that the client wants a version we support
 103         if e := c.checkClientVersions(h); e != nil {
 104                 return c, e
 105         }
 106         // Optional logging from connection start
 107         ln := senv.WantLogger()
 108         if ln != "" {
 109                 c.SetLogger(log.New(os.Stdout, ln+" ",
 110                         log.Ldate|log.Lmicroseconds|log.Lshortfile))
 111         }
 112
 113         // OK, put a CONNECT on the wire
 114         c.wtr = bufio.NewWriter(n) // Create the writer
 115         go c.writer()              // Start it
 116         var f Frame
 117         if senv.UseStomp() {
 118                 if ch.Value("accept-version") == SPL_11 || ch.Value("accept-version") == SPL_12 {
 119                         f = Frame{STOMP, ch, NULLBUFF} // Create actual STOMP frame
 120                 } else {
 121                         f = Frame{CONNECT, ch, NULLBUFF} // Create actual STOMP frame
 122                 }
 123                 // fmt.Printf("Frame: %q\n", f)
 124         } else {
 125                 f = Frame{CONNECT, ch, NULLBUFF} // Create actual CONNECT frame
 126                 // fmt.Printf("Frame: %q\n", f)
 127         }
 128         r := make(chan error)                               // Make the error channel for a write
 129         if e := c.writeWireData(wiredata{f, r}); e != nil { // Send the CONNECT frame
 130                 return c, e
 131         }
 132         e := <-r // Retrieve any error
 133         //
 134         if e != nil {
 135                 c.sysAbort() // Shutdown,  we are done with errors
 136                 return c, e
 137         }
 138         //fmt.Printf("CONDB03\n")
 139         //
 140         e = c.connectHandler(ch)
 141         if e != nil {
 142                 c.sysAbort() // Shutdown ,  we are done with errors
 143                 return c, e
 144         }
 145         //fmt.Printf("CONDB04\n")
 146         // We are connected
 147         go c.reader()
 148         //
 149         return c, e
 150 }