connect_helpers.go

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