src/dma/vendor/github.com/streadway/amqp/write.go

   1 // Copyright (c) 2012, Sean Treadway, SoundCloud Ltd.
   2 // Use of this source code is governed by a BSD-style
   3 // license that can be found in the LICENSE file.
   4 // Source code and contact info at http://github.com/streadway/amqp
   5
   6 package amqp
   7
   8 import (
   9         "bufio"
  10         "bytes"
  11         "encoding/binary"
  12         "errors"
  13         "io"
  14         "math"
  15         "time"
  16 )
  17
  18 func (w *writer) WriteFrame(frame frame) (err error) {
  19         if err = frame.write(w.w); err != nil {
  20                 return
  21         }
  22
  23         if buf, ok := w.w.(*bufio.Writer); ok {
  24                 err = buf.Flush()
  25         }
  26
  27         return
  28 }
  29
  30 func (f *methodFrame) write(w io.Writer) (err error) {
  31         var payload bytes.Buffer
  32
  33         if f.Method == nil {
  34                 return errors.New("malformed frame: missing method")
  35         }
  36
  37         class, method := f.Method.id()
  38
  39         if err = binary.Write(&payload, binary.BigEndian, class); err != nil {
  40                 return
  41         }
  42
  43         if err = binary.Write(&payload, binary.BigEndian, method); err != nil {
  44                 return
  45         }
  46
  47         if err = f.Method.write(&payload); err != nil {
  48                 return
  49         }
  50
  51         return writeFrame(w, frameMethod, f.ChannelId, payload.Bytes())
  52 }
  53
  54 // Heartbeat
  55 //
  56 // Payload is empty
  57 func (f *heartbeatFrame) write(w io.Writer) (err error) {
  58         return writeFrame(w, frameHeartbeat, f.ChannelId, []byte{})
  59 }
  60
  61 // CONTENT HEADER
  62 // 0          2        4           12               14
  63 // +----------+--------+-----------+----------------+------------- - -
  64 // | class-id | weight | body size | property flags | property list...
  65 // +----------+--------+-----------+----------------+------------- - -
  66 //    short     short    long long       short        remainder...
  67 //
  68 func (f *headerFrame) write(w io.Writer) (err error) {
  69         var payload bytes.Buffer
  70         var zeroTime time.Time
  71
  72         if err = binary.Write(&payload, binary.BigEndian, f.ClassId); err != nil {
  73                 return
  74         }
  75
  76         if err = binary.Write(&payload, binary.BigEndian, f.weight); err != nil {
  77                 return
  78         }
  79
  80         if err = binary.Write(&payload, binary.BigEndian, f.Size); err != nil {
  81                 return
  82         }
  83
  84         // First pass will build the mask to be serialized, second pass will serialize
  85         // each of the fields that appear in the mask.
  86
  87         var mask uint16
  88
  89         if len(f.Properties.ContentType) > 0 {
  90                 mask = mask | flagContentType
  91         }
  92         if len(f.Properties.ContentEncoding) > 0 {
  93                 mask = mask | flagContentEncoding
  94         }
  95         if f.Properties.Headers != nil && len(f.Properties.Headers) > 0 {
  96                 mask = mask | flagHeaders
  97         }
  98         if f.Properties.DeliveryMode > 0 {
  99                 mask = mask | flagDeliveryMode
 100         }
 101         if f.Properties.Priority > 0 {
 102                 mask = mask | flagPriority
 103         }
 104         if len(f.Properties.CorrelationId) > 0 {
 105                 mask = mask | flagCorrelationId
 106         }
 107         if len(f.Properties.ReplyTo) > 0 {
 108                 mask = mask | flagReplyTo
 109         }
 110         if len(f.Properties.Expiration) > 0 {
 111                 mask = mask | flagExpiration
 112         }
 113         if len(f.Properties.MessageId) > 0 {
 114                 mask = mask | flagMessageId
 115         }
 116         if f.Properties.Timestamp != zeroTime {
 117                 mask = mask | flagTimestamp
 118         }
 119         if len(f.Properties.Type) > 0 {
 120                 mask = mask | flagType
 121         }
 122         if len(f.Properties.UserId) > 0 {
 123                 mask = mask | flagUserId
 124         }
 125         if len(f.Properties.AppId) > 0 {
 126                 mask = mask | flagAppId
 127         }
 128
 129         if err = binary.Write(&payload, binary.BigEndian, mask); err != nil {
 130                 return
 131         }
 132
 133         if hasProperty(mask, flagContentType) {
 134                 if err = writeShortstr(&payload, f.Properties.ContentType); err != nil {
 135                         return
 136                 }
 137         }
 138         if hasProperty(mask, flagContentEncoding) {
 139                 if err = writeShortstr(&payload, f.Properties.ContentEncoding); err != nil {
 140                         return
 141                 }
 142         }
 143         if hasProperty(mask, flagHeaders) {
 144                 if err = writeTable(&payload, f.Properties.Headers); err != nil {
 145                         return
 146                 }
 147         }
 148         if hasProperty(mask, flagDeliveryMode) {
 149                 if err = binary.Write(&payload, binary.BigEndian, f.Properties.DeliveryMode); err != nil {
 150                         return
 151                 }
 152         }
 153         if hasProperty(mask, flagPriority) {
 154                 if err = binary.Write(&payload, binary.BigEndian, f.Properties.Priority); err != nil {
 155                         return
 156                 }
 157         }
 158         if hasProperty(mask, flagCorrelationId) {
 159                 if err = writeShortstr(&payload, f.Properties.CorrelationId); err != nil {
 160                         return
 161                 }
 162         }
 163         if hasProperty(mask, flagReplyTo) {
 164                 if err = writeShortstr(&payload, f.Properties.ReplyTo); err != nil {
 165                         return
 166                 }
 167         }
 168         if hasProperty(mask, flagExpiration) {
 169                 if err = writeShortstr(&payload, f.Properties.Expiration); err != nil {
 170                         return
 171                 }
 172         }
 173         if hasProperty(mask, flagMessageId) {
 174                 if err = writeShortstr(&payload, f.Properties.MessageId); err != nil {
 175                         return
 176                 }
 177         }
 178         if hasProperty(mask, flagTimestamp) {
 179                 if err = binary.Write(&payload, binary.BigEndian, uint64(f.Properties.Timestamp.Unix())); err != nil {
 180                         return
 181                 }
 182         }
 183         if hasProperty(mask, flagType) {
 184                 if err = writeShortstr(&payload, f.Properties.Type); err != nil {
 185                         return
 186                 }
 187         }
 188         if hasProperty(mask, flagUserId) {
 189                 if err = writeShortstr(&payload, f.Properties.UserId); err != nil {
 190                         return
 191                 }
 192         }
 193         if hasProperty(mask, flagAppId) {
 194                 if err = writeShortstr(&payload, f.Properties.AppId); err != nil {
 195                         return
 196                 }
 197         }
 198
 199         return writeFrame(w, frameHeader, f.ChannelId, payload.Bytes())
 200 }
 201
 202 // Body
 203 //
 204 // Payload is one byterange from the full body who's size is declared in the
 205 // Header frame
 206 func (f *bodyFrame) write(w io.Writer) (err error) {
 207         return writeFrame(w, frameBody, f.ChannelId, f.Body)
 208 }
 209
 210 func writeFrame(w io.Writer, typ uint8, channel uint16, payload []byte) (err error) {
 211         end := []byte{frameEnd}
 212         size := uint(len(payload))
 213
 214         _, err = w.Write([]byte{
 215                 byte(typ),
 216                 byte((channel & 0xff00) >> 8),
 217                 byte((channel & 0x00ff) >> 0),
 218                 byte((size & 0xff000000) >> 24),
 219                 byte((size & 0x00ff0000) >> 16),
 220                 byte((size & 0x0000ff00) >> 8),
 221                 byte((size & 0x000000ff) >> 0),
 222         })
 223
 224         if err != nil {
 225                 return
 226         }
 227
 228         if _, err = w.Write(payload); err != nil {
 229                 return
 230         }
 231
 232         if _, err = w.Write(end); err != nil {
 233                 return
 234         }
 235
 236         return
 237 }
 238
 239 func writeShortstr(w io.Writer, s string) (err error) {
 240         b := []byte(s)
 241
 242         var length = uint8(len(b))
 243
 244         if err = binary.Write(w, binary.BigEndian, length); err != nil {
 245                 return
 246         }
 247
 248         if _, err = w.Write(b[:length]); err != nil {
 249                 return
 250         }
 251
 252         return
 253 }
 254
 255 func writeLongstr(w io.Writer, s string) (err error) {
 256         b := []byte(s)
 257
 258         var length = uint32(len(b))
 259
 260         if err = binary.Write(w, binary.BigEndian, length); err != nil {
 261                 return
 262         }
 263
 264         if _, err = w.Write(b[:length]); err != nil {
 265                 return
 266         }
 267
 268         return
 269 }
 270
 271 /*
 272 'A': []interface{}
 273 'D': Decimal
 274 'F': Table
 275 'I': int32
 276 'S': string
 277 'T': time.Time
 278 'V': nil
 279 'b': byte
 280 'd': float64
 281 'f': float32
 282 'l': int64
 283 's': int16
 284 't': bool
 285 'x': []byte
 286 */
 287 func writeField(w io.Writer, value interface{}) (err error) {
 288         var buf [9]byte
 289         var enc []byte
 290
 291         switch v := value.(type) {
 292         case bool:
 293                 buf[0] = 't'
 294                 if v {
 295                         buf[1] = byte(1)
 296                 } else {
 297                         buf[1] = byte(0)
 298                 }
 299                 enc = buf[:2]
 300
 301         case byte:
 302                 buf[0] = 'b'
 303                 buf[1] = byte(v)
 304                 enc = buf[:2]
 305
 306         case int16:
 307                 buf[0] = 's'
 308                 binary.BigEndian.PutUint16(buf[1:3], uint16(v))
 309                 enc = buf[:3]
 310
 311         case int:
 312                 buf[0] = 'I'
 313                 binary.BigEndian.PutUint32(buf[1:5], uint32(v))
 314                 enc = buf[:5]
 315
 316         case int32:
 317                 buf[0] = 'I'
 318                 binary.BigEndian.PutUint32(buf[1:5], uint32(v))
 319                 enc = buf[:5]
 320
 321         case int64:
 322                 buf[0] = 'l'
 323                 binary.BigEndian.PutUint64(buf[1:9], uint64(v))
 324                 enc = buf[:9]
 325
 326         case float32:
 327                 buf[0] = 'f'
 328                 binary.BigEndian.PutUint32(buf[1:5], math.Float32bits(v))
 329                 enc = buf[:5]
 330
 331         case float64:
 332                 buf[0] = 'd'
 333                 binary.BigEndian.PutUint64(buf[1:9], math.Float64bits(v))
 334                 enc = buf[:9]
 335
 336         case Decimal:
 337                 buf[0] = 'D'
 338                 buf[1] = byte(v.Scale)
 339                 binary.BigEndian.PutUint32(buf[2:6], uint32(v.Value))
 340                 enc = buf[:6]
 341
 342         case string:
 343                 buf[0] = 'S'
 344                 binary.BigEndian.PutUint32(buf[1:5], uint32(len(v)))
 345                 enc = append(buf[:5], []byte(v)...)
 346
 347         case []interface{}: // field-array
 348                 buf[0] = 'A'
 349
 350                 sec := new(bytes.Buffer)
 351                 for _, val := range v {
 352                         if err = writeField(sec, val); err != nil {
 353                                 return
 354                         }
 355                 }
 356
 357                 binary.BigEndian.PutUint32(buf[1:5], uint32(sec.Len()))
 358                 if _, err = w.Write(buf[:5]); err != nil {
 359                         return
 360                 }
 361
 362                 if _, err = w.Write(sec.Bytes()); err != nil {
 363                         return
 364                 }
 365
 366                 return
 367
 368         case time.Time:
 369                 buf[0] = 'T'
 370                 binary.BigEndian.PutUint64(buf[1:9], uint64(v.Unix()))
 371                 enc = buf[:9]
 372
 373         case Table:
 374                 if _, err = w.Write([]byte{'F'}); err != nil {
 375                         return
 376                 }
 377                 return writeTable(w, v)
 378
 379         case []byte:
 380                 buf[0] = 'x'
 381                 binary.BigEndian.PutUint32(buf[1:5], uint32(len(v)))
 382                 if _, err = w.Write(buf[0:5]); err != nil {
 383                         return
 384                 }
 385                 if _, err = w.Write(v); err != nil {
 386                         return
 387                 }
 388                 return
 389
 390         case nil:
 391                 buf[0] = 'V'
 392                 enc = buf[:1]
 393
 394         default:
 395                 return ErrFieldType
 396         }
 397
 398         _, err = w.Write(enc)
 399
 400         return
 401 }
 402
 403 func writeTable(w io.Writer, table Table) (err error) {
 404         var buf bytes.Buffer
 405
 406         for key, val := range table {
 407                 if err = writeShortstr(&buf, key); err != nil {
 408                         return
 409                 }
 410                 if err = writeField(&buf, val); err != nil {
 411                         return
 412                 }
 413         }
 414
 415         return writeLongstr(w, string(buf.Bytes()))
 416 }