writer.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         "bytes"
  22         "net"
  23
  24         // "bytes"
  25         "strconv"
  26         "time"
  27 )
  28
  29 /*
  30         Write data to logical network writer.  Writer will take care of the output wire data.
  31         If the underlying connection goes bad and writer give up working, the closed ssdc chan
  32         will make sure write action aware that happens.
  33 */
  34 func (c *Connection) writeWireData(wd wiredata) error {
  35         select {
  36         case c.output <- wd:
  37         case <-c.ssdc:
  38                 return ECONBAD
  39         }
  40         return nil
  41 }
  42
  43 /*
  44         Logical network writer.  Read wiredata structures from the communication
  45         channel, and put the frame on the wire.
  46 */
  47 func (c *Connection) writer() {
  48 writerLoop:
  49         for {
  50                 select {
  51                 case d := <-c.output:
  52                         c.log("WTR_WIREWRITE start")
  53                         c.wireWrite(d)
  54                         logLock.Lock()
  55                         if c.logger != nil {
  56                                 c.logx("WTR_WIREWRITE COMPLETE", d.frame.Command, d.frame.Headers,
  57                                         HexData(d.frame.Body))
  58                         }
  59                         logLock.Unlock()
  60                         if d.frame.Command == DISCONNECT {
  61                                 break writerLoop // we are done with this connection
  62                         }
  63                 case _ = <-c.ssdc:
  64                         c.log("WTR_WIREWRITE shutdown S received")
  65                         break writerLoop
  66                 case _ = <-c.wtrsdc:
  67                         c.log("WTR_WIREWRITE shutdown W received")
  68                         break writerLoop
  69                 }
  70         } // of for
  71         //
  72         c.setConnected(false)
  73         c.sysAbort()
  74         c.log("WTR_SHUTDOWN", time.Now())
  75 }
  76
  77 /*
  78         Connection logical write.
  79 */
  80 func (c *Connection) wireWrite(d wiredata) {
  81         f := &d.frame
  82         // fmt.Printf("WWD01 f:[%v]\n", f)
  83         switch f.Command {
  84         case "\n": // HeartBeat frame
  85                 if c.dld.wde && c.dld.wds {
  86                         _ = c.netconn.SetWriteDeadline(time.Now().Add(c.dld.wdld))
  87                 }
  88                 _, e := c.wtr.WriteString(f.Command)
  89                 if e != nil {
  90                         if e.(net.Error).Timeout() {
  91                                 if c.dld.dns {
  92                                         c.dld.dlnotify(e, true)
  93                                 }
  94                         }
  95                         d.errchan <- e
  96                         return
  97                 }
  98         default: // Other frames
  99                 if e := f.writeFrame(c.wtr, c); e != nil {
 100                         d.errchan <- e
 101                         return
 102                 }
 103                 if e := c.wtr.Flush(); e != nil {
 104                         d.errchan <- e
 105                         return
 106                 }
 107         }
 108         if e := c.wtr.Flush(); e != nil {
 109                 d.errchan <- e
 110                 return
 111         }
 112         //
 113         if c.hbd != nil {
 114                 c.hbd.sdl.Lock()
 115                 c.hbd.ls = time.Now().UnixNano() // Latest good send
 116                 c.hbd.sdl.Unlock()
 117         }
 118         c.mets.tfw++                // Frame written count
 119         c.mets.tbw += f.Size(false) // Bytes written count
 120         //
 121         d.errchan <- nil
 122         return
 123 }
 124
 125 /*
 126         Physical frame write to the wire.
 127 */
 128 func (f *Frame) writeFrame(w *bufio.Writer, c *Connection) error {
 129
 130         var sctok bool
 131         // Content type.  Always add it if the client does not suppress and does not
 132         // supply it.
 133         _, sctok = f.Headers.Contains(HK_SUPPRESS_CT)
 134         if !sctok {
 135                 if _, ctok := f.Headers.Contains(HK_CONTENT_TYPE); !ctok {
 136                         f.Headers = append(f.Headers, HK_CONTENT_TYPE,
 137                                 DFLT_CONTENT_TYPE)
 138                 }
 139         }
 140
 141         var sclok bool
 142         // Content length - Always add it if client does not suppress it and
 143         // does not supply it.
 144         _, sclok = f.Headers.Contains(HK_SUPPRESS_CL)
 145         if !sclok {
 146                 if _, clok := f.Headers.Contains(HK_CONTENT_LENGTH); !clok {
 147                         f.Headers = append(f.Headers, HK_CONTENT_LENGTH, strconv.Itoa(len(f.Body)))
 148                 }
 149         }
 150         // Encode the headers if needed
 151         if c.Protocol() > SPL_10 && f.Command != CONNECT {
 152                 for i := 0; i < len(f.Headers); i += 2 {
 153                         f.Headers[i] = encode(f.Headers[i])
 154                         f.Headers[i+1] = encode(f.Headers[i+1])
 155                 }
 156         }
 157
 158         if sclok {
 159                 nz := bytes.IndexByte(f.Body, 0)
 160                 // fmt.Printf("WDBG41 ok:%v\n", nz)
 161                 if nz == 0 {
 162                         f.Body = []byte{}
 163                         // fmt.Printf("WDBG42 body:%v bodystring: %v\n", f.Body, string(f.Body))
 164                 } else if nz > 0 {
 165                         f.Body = f.Body[0:nz]
 166                         // fmt.Printf("WDBG43 body:%v bodystring: %v\n", f.Body, string(f.Body))
 167                 }
 168         }
 169
 170         if c.dld.wde && c.dld.wds {
 171                 _ = c.netconn.SetWriteDeadline(time.Now().Add(c.dld.wdld))
 172         }
 173
 174         // Writes start
 175
 176         // Write the frame Command
 177         _, e := w.WriteString(f.Command + "\n")
 178         if c.checkWriteError(e) != nil {
 179                 return e
 180         }
 181         // fmt.Println("WRCMD", f.Command)
 182         // Write the frame Headers
 183         for i := 0; i < len(f.Headers); i += 2 {
 184                 if c.dld.wde && c.dld.wds {
 185                         _ = c.netconn.SetWriteDeadline(time.Now().Add(c.dld.wdld))
 186                 }
 187                 _, e := w.WriteString(f.Headers[i] + ":" + f.Headers[i+1] + "\n")
 188                 if c.checkWriteError(e) != nil {
 189                         return e
 190                 }
 191                 // fmt.Println("WRHDR", f.Headers[i]+":"+f.Headers[i+1]+"\n")
 192         }
 193
 194         // Write the last Header LF
 195         if c.dld.wde && c.dld.wds {
 196                 _ = c.netconn.SetWriteDeadline(time.Now().Add(c.dld.wdld))
 197         }
 198         e = w.WriteByte('\n')
 199         if c.checkWriteError(e) != nil {
 200                 return e
 201         }
 202         // fmt.Printf("WDBG40 ok:%v\n", sclok)
 203
 204         // Write the body
 205         if len(f.Body) != 0 { // Foolish to write 0 length data
 206                 // fmt.Println("WRBDY", f.Body)
 207                 e := c.writeBody(f)
 208                 if c.checkWriteError(e) != nil {
 209                         return e
 210                 }
 211         }
 212         if c.dld.wde && c.dld.wds {
 213                 _ = c.netconn.SetWriteDeadline(time.Now().Add(c.dld.wdld))
 214         }
 215         e = w.WriteByte(0)
 216         if c.checkWriteError(e) != nil {
 217                 return e
 218         }
 219         // End of write loop - set no deadline
 220         if c.dld.wde {
 221                 _ = c.netconn.SetWriteDeadline(c.dld.t0)
 222         }
 223         return nil
 224 }
 225
 226 func (c *Connection) checkWriteError(e error) error {
 227         if e == nil {
 228                 return e
 229         }
 230         ne, ok := e.(net.Error)
 231         if !ok {
 232                 return e
 233         }
 234         if ne.Timeout() {
 235                 if c.dld.dns {
 236                         c.log("invoking write deadline callback 1")
 237                         c.dld.dlnotify(e, true)
 238                 }
 239         }
 240         return e
 241 }
 242
 243 func (c *Connection) writeBody(f *Frame) error {
 244         // fmt.Printf("WDBG99 body:%v bodystring: %v\n", f.Body, string(f.Body))
 245         var n = 0
 246         var e error
 247         for {
 248                 if c.dld.wde && c.dld.wds {
 249                         _ = c.netconn.SetWriteDeadline(time.Now().Add(c.dld.wdld))
 250                 }
 251                 n, e = c.wtr.Write(f.Body)
 252                 if n == len(f.Body) {
 253                         return e
 254                 }
 255                 c.log("SHORT WRITE", n, len(f.Body))
 256                 if n == 0 { // Zero bytes would mean something is seriously wrong.
 257                         return e
 258                 }
 259                 if !c.dld.rfsw {
 260                         return e
 261                 }
 262                 if c.dld.wde && c.dld.wds && c.dld.dns && isErrorTimeout(e) {
 263                         c.log("invoking write deadline callback 2")
 264                         c.dld.dlnotify(e, true)
 265                 }
 266                 // *Any* error from a bufio.Writer is *not* recoverable.  See code in
 267                 // bufio.go to understand this.  We get a new writer here, to clear any
 268                 // error condition.
 269                 c.wtr = bufio.NewWriter(c.netconn) // Create new writer
 270                 f.Body = f.Body[n:]
 271         }
 272 }
 273
 274 func isErrorTimeout(e error) bool {
 275         if e == nil {
 276                 return false
 277         }
 278         _, ok := e.(net.Error)
 279         if !ok {
 280                 return false
 281         }
 282         return true
 283 }