connect_helpers.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         "bytes"
  22
  23         // "fmt"
  24         "strings"
  25 )
  26
  27 type CONNERROR struct {
  28         err  error
  29         desc string
  30 }
  31
  32 func (e *CONNERROR) Error() string {
  33         return e.err.Error() + ":" + e.desc
  34 }
  35
  36 /*
  37         Connection handler, one time use during initial connect.
  38
  39         Handle broker response, react to version incompatabilities, set up session,
  40         and if necessary initialize heart beats.
  41 */
  42 func (c *Connection) connectHandler(h Headers) (e error) {
  43         //fmt.Printf("CHDB01\n")
  44         c.rdr = bufio.NewReader(c.netconn)
  45         b, e := c.rdr.ReadBytes(0)
  46         if e != nil {
  47                 return e
  48         }
  49         //fmt.Printf("CHDB02\n")
  50         f, e := connectResponse(string(b))
  51         if e != nil {
  52                 return e
  53         }
  54         //fmt.Printf("CHDB03\n")
  55         //
  56         c.ConnectResponse = &Message{f.Command, f.Headers, f.Body}
  57         if c.ConnectResponse.Command == ERROR {
  58                 return &CONNERROR{ECONERR, string(f.Body)}
  59         }
  60         //fmt.Printf("CHDB04\n")
  61         //
  62         e = c.setProtocolLevel(h, c.ConnectResponse.Headers)
  63         if e != nil {
  64                 return e
  65         }
  66         //fmt.Printf("CHDB05\n")
  67         //
  68         if s, ok := c.ConnectResponse.Headers.Contains(HK_SESSION); ok {
  69                 c.sessLock.Lock()
  70                 c.session = s
  71                 c.sessLock.Unlock()
  72         }
  73
  74         if c.Protocol() >= SPL_11 {
  75                 e = c.initializeHeartBeats(h)
  76                 if e != nil {
  77                         return e
  78                 }
  79         }
  80         //fmt.Printf("CHDB06\n")
  81
  82         c.connected = true
  83         c.mets.tfr += 1
  84         c.mets.tbr += c.ConnectResponse.Size(false)
  85         return nil
  86 }
  87
  88 /*
  89         Handle data from the wire after CONNECT is sent. Attempt to create a Frame
  90         from the wire data.
  91
  92         Called one time per connection at connection start.
  93 */
  94 func connectResponse(s string) (*Frame, error) {
  95         //
  96         f := new(Frame)
  97         f.Headers = Headers{}
  98         f.Body = make([]uint8, 0)
  99
 100         // Get f.Command
 101         c := strings.SplitN(s, "\n", 2)
 102         if len(c) < 2 {
 103                 if len(c) == 1 {
 104                         // fmt.Printf("lenc is: %d, data:%#v\n", len(c), c[0])
 105                         if bytes.Compare(HandShake, []byte(c[0])) == 0 {
 106                                 return nil, EBADSSLP
 107                         }
 108                 }
 109                 return nil, EBADFRM
 110         }
 111         f.Command = c[0]
 112         if f.Command != CONNECTED && f.Command != ERROR {
 113                 return f, EUNKFRM
 114         }
 115
 116         switch c[1] {
 117         case "\x00", "\n": // No headers, malformed bodies
 118                 f.Body = []uint8(c[1])
 119                 return f, EBADFRM
 120         case "\n\x00": // No headers, no body is OK
 121                 return f, nil
 122         default: // Otherwise continue
 123         }
 124
 125         b := strings.SplitN(c[1], "\n\n", 2)
 126         if len(b) == 1 { // No Headers, b[0] == body
 127                 w := []uint8(b[0])
 128                 f.Body = w[0 : len(w)-1]
 129                 if f.Command == CONNECTED && len(f.Body) > 0 {
 130                         return f, EBDYDATA
 131                 }
 132                 return f, nil
 133         }
 134
 135         // Here:
 136         // b[0] - the headers
 137         // b[1] - the body
 138
 139         // Get f.Headers
 140         for _, l := range strings.Split(b[0], "\n") {
 141                 p := strings.SplitN(l, ":", 2)
 142                 if len(p) < 2 {
 143                         f.Body = []uint8(p[0]) // Bad feedback
 144                         return f, EUNKHDR
 145                 }
 146                 f.Headers = append(f.Headers, p[0], p[1])
 147         }
 148         // get f.Body
 149         w := []uint8(b[1])
 150         f.Body = w[0 : len(w)-1]
 151         if f.Command == CONNECTED && len(f.Body) > 0 {
 152                 return f, EBDYDATA
 153         }
 154
 155         return f, nil
 156 }
 157
 158 /*
 159         Check client version, one time use during initial connect.
 160 */
 161 func (c *Connection) checkClientVersions(h Headers) (e error) {
 162         w := h.Value(HK_ACCEPT_VERSION)
 163         if w == "" { // Not present, client wants 1.0
 164                 return nil
 165         }
 166         v := strings.SplitN(w, ",", -1) //
 167         ok := false
 168         for _, sv := range v {
 169                 if hasValue(supported, sv) {
 170                         ok = true // At least we support one the client wants
 171                 }
 172         }
 173         if !ok {
 174                 return EBADVERCLI
 175         }
 176         if _, ok = h.Contains(HK_HOST); !ok {
 177                 return EREQHOST
 178         }
 179         return nil
 180 }
 181
 182 /*
 183         Set the protocol level for this new connection.
 184 */
 185 func (c *Connection) setProtocolLevel(ch, sh Headers) (e error) {
 186         chw := ch.Value(HK_ACCEPT_VERSION)
 187         shr := sh.Value(HK_VERSION)
 188
 189         if chw == shr && Supported(shr) {
 190                 c.protocol = shr
 191                 return nil
 192         }
 193         if chw == "" && shr == "" { // Straight up 1.0
 194                 return nil // protocol level defaults to SPL_10
 195         }
 196         cv := strings.SplitN(chw, ",", -1) // Client requested versions
 197
 198         if chw != "" && shr != "" {
 199                 if hasValue(cv, shr) {
 200                         if !Supported(shr) {
 201                                 return EBADVERSVR // Client and server agree, but we do not support it
 202                         }
 203                         c.protocol = shr
 204                         return nil
 205                 } else {
 206                         return EBADVERCLI
 207                 }
 208         }
 209         if chw != "" && shr == "" { // Client asked for something, server is pure 1.0
 210                 if hasValue(cv, SPL_10) {
 211                         return nil // protocol level defaults to SPL_10
 212                 }
 213         }
 214
 215         c.protocol = shr // Could be anything we support
 216         return nil
 217 }
 218
 219 /*
 220         Internal function, used only during CONNECT processing.
 221 */
 222 func hasValue(a []string, w string) bool {
 223         for _, v := range a {
 224                 if v == w {
 225                         return true
 226                 }
 227         }
 228         return false
 229 }