connect.go

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