mirror of https://github.com/tidwall/tile38.git
457 lines
9.3 KiB
Go
457 lines
9.3 KiB
Go
// Copyright (c) 2012, Sean Treadway, SoundCloud Ltd.
|
|
// Use of this source code is governed by a BSD-style
|
|
// license that can be found in the LICENSE file.
|
|
// Source code and contact info at http://github.com/streadway/amqp
|
|
|
|
package amqp
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/binary"
|
|
"errors"
|
|
"io"
|
|
"time"
|
|
)
|
|
|
|
/*
|
|
Reads a frame from an input stream and returns an interface that can be cast into
|
|
one of the following:
|
|
|
|
methodFrame
|
|
PropertiesFrame
|
|
bodyFrame
|
|
heartbeatFrame
|
|
|
|
2.3.5 frame Details
|
|
|
|
All frames consist of a header (7 octets), a payload of arbitrary size, and a
|
|
'frame-end' octet that detects malformed frames:
|
|
|
|
0 1 3 7 size+7 size+8
|
|
+------+---------+-------------+ +------------+ +-----------+
|
|
| type | channel | size | | payload | | frame-end |
|
|
+------+---------+-------------+ +------------+ +-----------+
|
|
octet short long size octets octet
|
|
|
|
To read a frame, we:
|
|
1. Read the header and check the frame type and channel.
|
|
2. Depending on the frame type, we read the payload and process it.
|
|
3. Read the frame end octet.
|
|
|
|
In realistic implementations where performance is a concern, we would use
|
|
“read-ahead buffering” or
|
|
|
|
“gathering reads” to avoid doing three separate system calls to read a frame.
|
|
*/
|
|
func (r *reader) ReadFrame() (frame frame, err error) {
|
|
var scratch [7]byte
|
|
|
|
if _, err = io.ReadFull(r.r, scratch[:7]); err != nil {
|
|
return
|
|
}
|
|
|
|
typ := uint8(scratch[0])
|
|
channel := binary.BigEndian.Uint16(scratch[1:3])
|
|
size := binary.BigEndian.Uint32(scratch[3:7])
|
|
|
|
switch typ {
|
|
case frameMethod:
|
|
if frame, err = r.parseMethodFrame(channel, size); err != nil {
|
|
return
|
|
}
|
|
|
|
case frameHeader:
|
|
if frame, err = r.parseHeaderFrame(channel, size); err != nil {
|
|
return
|
|
}
|
|
|
|
case frameBody:
|
|
if frame, err = r.parseBodyFrame(channel, size); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
case frameHeartbeat:
|
|
if frame, err = r.parseHeartbeatFrame(channel, size); err != nil {
|
|
return
|
|
}
|
|
|
|
default:
|
|
return nil, ErrFrame
|
|
}
|
|
|
|
if _, err = io.ReadFull(r.r, scratch[:1]); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if scratch[0] != frameEnd {
|
|
return nil, ErrFrame
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
func readShortstr(r io.Reader) (v string, err error) {
|
|
var length uint8
|
|
if err = binary.Read(r, binary.BigEndian, &length); err != nil {
|
|
return
|
|
}
|
|
|
|
bytes := make([]byte, length)
|
|
if _, err = io.ReadFull(r, bytes); err != nil {
|
|
return
|
|
}
|
|
return string(bytes), nil
|
|
}
|
|
|
|
func readLongstr(r io.Reader) (v string, err error) {
|
|
var length uint32
|
|
if err = binary.Read(r, binary.BigEndian, &length); err != nil {
|
|
return
|
|
}
|
|
|
|
// slices can't be longer than max int32 value
|
|
if length > (^uint32(0) >> 1) {
|
|
return
|
|
}
|
|
|
|
bytes := make([]byte, length)
|
|
if _, err = io.ReadFull(r, bytes); err != nil {
|
|
return
|
|
}
|
|
return string(bytes), nil
|
|
}
|
|
|
|
func readDecimal(r io.Reader) (v Decimal, err error) {
|
|
if err = binary.Read(r, binary.BigEndian, &v.Scale); err != nil {
|
|
return
|
|
}
|
|
if err = binary.Read(r, binary.BigEndian, &v.Value); err != nil {
|
|
return
|
|
}
|
|
return
|
|
}
|
|
|
|
func readFloat32(r io.Reader) (v float32, err error) {
|
|
if err = binary.Read(r, binary.BigEndian, &v); err != nil {
|
|
return
|
|
}
|
|
return
|
|
}
|
|
|
|
func readFloat64(r io.Reader) (v float64, err error) {
|
|
if err = binary.Read(r, binary.BigEndian, &v); err != nil {
|
|
return
|
|
}
|
|
return
|
|
}
|
|
|
|
func readTimestamp(r io.Reader) (v time.Time, err error) {
|
|
var sec int64
|
|
if err = binary.Read(r, binary.BigEndian, &sec); err != nil {
|
|
return
|
|
}
|
|
return time.Unix(sec, 0), nil
|
|
}
|
|
|
|
/*
|
|
'A': []interface{}
|
|
'D': Decimal
|
|
'F': Table
|
|
'I': int32
|
|
'S': string
|
|
'T': time.Time
|
|
'V': nil
|
|
'b': byte
|
|
'd': float64
|
|
'f': float32
|
|
'l': int64
|
|
's': int16
|
|
't': bool
|
|
'x': []byte
|
|
*/
|
|
func readField(r io.Reader) (v interface{}, err error) {
|
|
var typ byte
|
|
if err = binary.Read(r, binary.BigEndian, &typ); err != nil {
|
|
return
|
|
}
|
|
|
|
switch typ {
|
|
case 't':
|
|
var value uint8
|
|
if err = binary.Read(r, binary.BigEndian, &value); err != nil {
|
|
return
|
|
}
|
|
return (value != 0), nil
|
|
|
|
case 'b':
|
|
var value [1]byte
|
|
if _, err = io.ReadFull(r, value[0:1]); err != nil {
|
|
return
|
|
}
|
|
return value[0], nil
|
|
|
|
case 's':
|
|
var value int16
|
|
if err = binary.Read(r, binary.BigEndian, &value); err != nil {
|
|
return
|
|
}
|
|
return value, nil
|
|
|
|
case 'I':
|
|
var value int32
|
|
if err = binary.Read(r, binary.BigEndian, &value); err != nil {
|
|
return
|
|
}
|
|
return value, nil
|
|
|
|
case 'l':
|
|
var value int64
|
|
if err = binary.Read(r, binary.BigEndian, &value); err != nil {
|
|
return
|
|
}
|
|
return value, nil
|
|
|
|
case 'f':
|
|
var value float32
|
|
if err = binary.Read(r, binary.BigEndian, &value); err != nil {
|
|
return
|
|
}
|
|
return value, nil
|
|
|
|
case 'd':
|
|
var value float64
|
|
if err = binary.Read(r, binary.BigEndian, &value); err != nil {
|
|
return
|
|
}
|
|
return value, nil
|
|
|
|
case 'D':
|
|
return readDecimal(r)
|
|
|
|
case 'S':
|
|
return readLongstr(r)
|
|
|
|
case 'A':
|
|
return readArray(r)
|
|
|
|
case 'T':
|
|
return readTimestamp(r)
|
|
|
|
case 'F':
|
|
return readTable(r)
|
|
|
|
case 'x':
|
|
var len int32
|
|
if err = binary.Read(r, binary.BigEndian, &len); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
value := make([]byte, len)
|
|
if _, err = io.ReadFull(r, value); err != nil {
|
|
return nil, err
|
|
}
|
|
return value, err
|
|
|
|
case 'V':
|
|
return nil, nil
|
|
}
|
|
|
|
return nil, ErrSyntax
|
|
}
|
|
|
|
/*
|
|
Field tables are long strings that contain packed name-value pairs. The
|
|
name-value pairs are encoded as short string defining the name, and octet
|
|
defining the values type and then the value itself. The valid field types for
|
|
tables are an extension of the native integer, bit, string, and timestamp
|
|
types, and are shown in the grammar. Multi-octet integer fields are always
|
|
held in network byte order.
|
|
*/
|
|
func readTable(r io.Reader) (table Table, err error) {
|
|
var nested bytes.Buffer
|
|
var str string
|
|
|
|
if str, err = readLongstr(r); err != nil {
|
|
return
|
|
}
|
|
|
|
nested.Write([]byte(str))
|
|
|
|
table = make(Table)
|
|
|
|
for nested.Len() > 0 {
|
|
var key string
|
|
var value interface{}
|
|
|
|
if key, err = readShortstr(&nested); err != nil {
|
|
return
|
|
}
|
|
|
|
if value, err = readField(&nested); err != nil {
|
|
return
|
|
}
|
|
|
|
table[key] = value
|
|
}
|
|
|
|
return
|
|
}
|
|
|
|
func readArray(r io.Reader) ([]interface{}, error) {
|
|
var (
|
|
size uint32
|
|
err error
|
|
)
|
|
|
|
if err = binary.Read(r, binary.BigEndian, &size); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var (
|
|
lim = &io.LimitedReader{R: r, N: int64(size)}
|
|
arr = []interface{}{}
|
|
field interface{}
|
|
)
|
|
|
|
for {
|
|
if field, err = readField(lim); err != nil {
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
return nil, err
|
|
}
|
|
arr = append(arr, field)
|
|
}
|
|
|
|
return arr, nil
|
|
}
|
|
|
|
// Checks if this bit mask matches the flags bitset
|
|
func hasProperty(mask uint16, prop int) bool {
|
|
return int(mask)&prop > 0
|
|
}
|
|
|
|
func (r *reader) parseHeaderFrame(channel uint16, size uint32) (frame frame, err error) {
|
|
hf := &headerFrame{
|
|
ChannelId: channel,
|
|
}
|
|
|
|
if err = binary.Read(r.r, binary.BigEndian, &hf.ClassId); err != nil {
|
|
return
|
|
}
|
|
|
|
if err = binary.Read(r.r, binary.BigEndian, &hf.weight); err != nil {
|
|
return
|
|
}
|
|
|
|
if err = binary.Read(r.r, binary.BigEndian, &hf.Size); err != nil {
|
|
return
|
|
}
|
|
|
|
var flags uint16
|
|
|
|
if err = binary.Read(r.r, binary.BigEndian, &flags); err != nil {
|
|
return
|
|
}
|
|
|
|
if hasProperty(flags, flagContentType) {
|
|
if hf.Properties.ContentType, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagContentEncoding) {
|
|
if hf.Properties.ContentEncoding, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagHeaders) {
|
|
if hf.Properties.Headers, err = readTable(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagDeliveryMode) {
|
|
if err = binary.Read(r.r, binary.BigEndian, &hf.Properties.DeliveryMode); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagPriority) {
|
|
if err = binary.Read(r.r, binary.BigEndian, &hf.Properties.Priority); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagCorrelationId) {
|
|
if hf.Properties.CorrelationId, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagReplyTo) {
|
|
if hf.Properties.ReplyTo, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagExpiration) {
|
|
if hf.Properties.Expiration, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagMessageId) {
|
|
if hf.Properties.MessageId, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagTimestamp) {
|
|
if hf.Properties.Timestamp, err = readTimestamp(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagType) {
|
|
if hf.Properties.Type, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagUserId) {
|
|
if hf.Properties.UserId, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagAppId) {
|
|
if hf.Properties.AppId, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if hasProperty(flags, flagReserved1) {
|
|
if hf.Properties.reserved1, err = readShortstr(r.r); err != nil {
|
|
return
|
|
}
|
|
}
|
|
|
|
return hf, nil
|
|
}
|
|
|
|
func (r *reader) parseBodyFrame(channel uint16, size uint32) (frame frame, err error) {
|
|
bf := &bodyFrame{
|
|
ChannelId: channel,
|
|
Body: make([]byte, size),
|
|
}
|
|
|
|
if _, err = io.ReadFull(r.r, bf.Body); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return bf, nil
|
|
}
|
|
|
|
var errHeartbeatPayload = errors.New("Heartbeats should not have a payload")
|
|
|
|
func (r *reader) parseHeartbeatFrame(channel uint16, size uint32) (frame frame, err error) {
|
|
hf := &heartbeatFrame{
|
|
ChannelId: channel,
|
|
}
|
|
|
|
if size > 0 {
|
|
return nil, errHeartbeatPayload
|
|
}
|
|
|
|
return hf, nil
|
|
}
|