Add unit tests for AMQP dissector (#806)

* Add unit tests for AMQP dissector

* Always use UTC as the timezone in the timestamps

* Remove the dead code

* Remove more dead code

* Fix `dissector.Analyze` call
Co-authored-by: gadotroee <55343099+gadotroee@users.noreply.github.com>

Makefile

@@ -103,6 +103,7 @@ test-shared:  ## Run shared tests
 
 test-extensions:  ## Run extensions tests
 	@echo "running http tests"; cd tap/extensions/http && $(MAKE) test
+	@echo "running amqp tests"; cd tap/extensions/amqp && $(MAKE) test
 
 acceptance-test:  ## Run acceptance tests
 	@echo "running acceptance tests"; cd acceptanceTests && $(MAKE) test

tap/extensions/amqp/Makefile

+skipbin := $$(find bin -mindepth 1 -maxdepth 1)
+skipexpect := $$(find expect -mindepth 1 -maxdepth 1)
+
+test: test-pull-bin test-pull-expect
+	@MIZU_TEST=1 go test -v ./... -coverpkg=./... -race -coverprofile=coverage.out -covermode=atomic
+
+test-update: test-pull-bin
+	@MIZU_TEST=1 TEST_UPDATE=1 go test -v ./... -coverpkg=./... -coverprofile=coverage.out -covermode=atomic
+
+test-pull-bin:
+	@mkdir -p bin
+	@[ "${skipbin}" ] && echo "Skipping downloading BINs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp gs://static.up9.io/mizu/test-pcap/bin/amqp/\*.bin bin
+
+test-pull-expect:
+	@mkdir -p expect
+	@[ "${skipexpect}" ] && echo "Skipping downloading expected JSONs" || gsutil  -o 'GSUtil:parallel_process_count=5' -o 'GSUtil:parallel_thread_count=5'  -m cp -r gs://static.up9.io/mizu/test-pcap/expect/amqp/\* expect

tap/extensions/amqp/go.mod

@@ -2,8 +2,16 @@ module github.com/up9inc/mizu/tap/extensions/amqp
 
 go 1.17
 
-require github.com/up9inc/mizu/tap/api v0.0.0
+require (
+	github.com/stretchr/testify v1.7.0
+	github.com/up9inc/mizu/tap/api v0.0.0
+)
 
-require github.com/google/martian v2.1.0+incompatible // indirect
+require (
+	github.com/davecgh/go-spew v1.1.0 // indirect
+	github.com/google/martian v2.1.0+incompatible // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect
+)
 
 replace github.com/up9inc/mizu/tap/api v0.0.0 => ../../api

tap/extensions/amqp/go.sum

+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/google/martian v2.1.0+incompatible h1:/CP5g8u/VJHijgedC/Legn3BAbAaWPgecwXBIDzw5no=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

tap/extensions/amqp/helpers.go

@@ -3,6 +3,7 @@ package amqp
 import (
 	"encoding/json"
 	"fmt"
+	"sort"
 	"strconv"
 	"time"
 
@@ -282,6 +283,9 @@ func representBasicPublish(event map[string]interface{}) []interface{} {
 				Selector: fmt.Sprintf(`request.properties.headers["%s"]`, name),
 			})
 		}
+		sort.Slice(headers, func(i, j int) bool {
+			return headers[i].Name < headers[j].Name
+		})
 		headersMarshaled, _ := json.Marshal(headers)
 		rep = append(rep, api.SectionData{
 			Type:  api.TABLE,
@@ -366,6 +370,9 @@ func representBasicDeliver(event map[string]interface{}) []interface{} {
 				Selector: fmt.Sprintf(`request.properties.headers["%s"]`, name),
 			})
 		}
+		sort.Slice(headers, func(i, j int) bool {
+			return headers[i].Name < headers[j].Name
+		})
 		headersMarshaled, _ := json.Marshal(headers)
 		rep = append(rep, api.SectionData{
 			Type:  api.TABLE,
@@ -438,6 +445,9 @@ func representQueueDeclare(event map[string]interface{}) []interface{} {
 				Selector: fmt.Sprintf(`request.arguments["%s"]`, name),
 			})
 		}
+		sort.Slice(headers, func(i, j int) bool {
+			return headers[i].Name < headers[j].Name
+		})
 		headersMarshaled, _ := json.Marshal(headers)
 		rep = append(rep, api.SectionData{
 			Type:  api.TABLE,
@@ -504,6 +514,9 @@ func representExchangeDeclare(event map[string]interface{}) []interface{} {
 				Selector: fmt.Sprintf(`request.arguments["%s"]`, name),
 			})
 		}
+		sort.Slice(headers, func(i, j int) bool {
+			return headers[i].Name < headers[j].Name
+		})
 		headersMarshaled, _ := json.Marshal(headers)
 		rep = append(rep, api.SectionData{
 			Type:  api.TABLE,
@@ -565,6 +578,9 @@ func representConnectionStart(event map[string]interface{}) []interface{} {
 				Selector: fmt.Sprintf(`request.serverProperties["%s"]`, name),
 			})
 		}
+		sort.Slice(headers, func(i, j int) bool {
+			return headers[i].Name < headers[j].Name
+		})
 		headersMarshaled, _ := json.Marshal(headers)
 		rep = append(rep, api.SectionData{
 			Type:  api.TABLE,
@@ -656,6 +672,9 @@ func representQueueBind(event map[string]interface{}) []interface{} {
 				Selector: fmt.Sprintf(`request.arguments["%s"]`, name),
 			})
 		}
+		sort.Slice(headers, func(i, j int) bool {
+			return headers[i].Name < headers[j].Name
+		})
 		headersMarshaled, _ := json.Marshal(headers)
 		rep = append(rep, api.SectionData{
 			Type:  api.TABLE,
@@ -717,6 +736,9 @@ func representBasicConsume(event map[string]interface{}) []interface{} {
 				Selector: fmt.Sprintf(`request.arguments["%s"]`, name),
 			})
 		}
+		sort.Slice(headers, func(i, j int) bool {
+			return headers[i].Name < headers[j].Name
+		})
 		headersMarshaled, _ := json.Marshal(headers)
 		rep = append(rep, api.SectionData{
 			Type:  api.TABLE,

tap/extensions/amqp/main.go

@@ -8,6 +8,7 @@ import (
 	"io"
 	"log"
 	"strconv"
+	"time"
 
 	"github.com/up9inc/mizu/tap/api"
 )
@@ -85,7 +86,7 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 		frame, err := r.ReadFrame()
 		if err == io.EOF {
 			// We must read until we see an EOF... very important!
-			return errors.New("AMQP EOF")
+			return nil
 		}
 
 		switch f := frame.(type) {
@@ -96,6 +97,12 @@ func (d dissecting) Dissect(b *bufio.Reader, isClient bool, tcpID *api.TcpID, co
 			// start content state
 			header = f
 			remaining = int(header.Size)
+
+			// Workaround for `Time.MarshalJSON: year outside of range [0,9999]` error
+			if header.Properties.Timestamp.Year() > 9999 {
+				header.Properties.Timestamp = time.Time{}.UTC()
+			}
+
 			switch lastMethodFrameMessage.(type) {
 			case *BasicPublish:
 				eventBasicPublish.Properties = header.Properties

tap/extensions/amqp/main_test.go

+package amqp
+
+import (
+	"bufio"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"log"
+	"os"
+	"path"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/up9inc/mizu/tap/api"
+)
+
+const (
+	binDir          = "bin"
+	patternBin      = "*_req.bin"
+	patternDissect  = "*.json"
+	msgDissecting   = "Dissecting:"
+	msgAnalyzing    = "Analyzing:"
+	msgRepresenting = "Representing:"
+	respSuffix      = "_res.bin"
+	expectDir       = "expect"
+	dissectDir      = "dissect"
+	analyzeDir      = "analyze"
+	representDir    = "represent"
+	testUpdate      = "TEST_UPDATE"
+)
+
+func TestRegister(t *testing.T) {
+	dissector := NewDissector()
+	extension := &api.Extension{}
+	dissector.Register(extension)
+	assert.Equal(t, "amqp", extension.Protocol.Name)
+}
+
+func TestMacros(t *testing.T) {
+	expectedMacros := map[string]string{
+		"amqp": `proto.name == "amqp"`,
+	}
+	dissector := NewDissector()
+	macros := dissector.Macros()
+	assert.Equal(t, expectedMacros, macros)
+}
+
+func TestPing(t *testing.T) {
+	dissector := NewDissector()
+	dissector.Ping()
+}
+
+func TestDissect(t *testing.T) {
+	_, testUpdateEnabled := os.LookupEnv(testUpdate)
+
+	expectDirDissect := path.Join(expectDir, dissectDir)
+
+	if testUpdateEnabled {
+		os.RemoveAll(expectDirDissect)
+		err := os.MkdirAll(expectDirDissect, 0775)
+		assert.Nil(t, err)
+	}
+
+	dissector := NewDissector()
+	paths, err := filepath.Glob(path.Join(binDir, patternBin))
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	options := &api.TrafficFilteringOptions{
+		IgnoredUserAgents: []string{},
+	}
+
+	for _, _path := range paths {
+		basePath := _path[:len(_path)-8]
+
+		// Channel to verify the output
+		itemChannel := make(chan *api.OutputChannelItem)
+		var emitter api.Emitter = &api.Emitting{
+			AppStats:      &api.AppStats{},
+			OutputChannel: itemChannel,
+		}
+
+		var items []*api.OutputChannelItem
+		stop := make(chan bool)
+
+		go func() {
+			for {
+				select {
+				case <-stop:
+					return
+				case item := <-itemChannel:
+					items = append(items, item)
+				}
+			}
+		}()
+
+		// Stream level
+		counterPair := &api.CounterPair{
+			Request:  0,
+			Response: 0,
+		}
+		superIdentifier := &api.SuperIdentifier{}
+
+		// Request
+		pathClient := _path
+		fmt.Printf("%s %s\n", msgDissecting, pathClient)
+		fileClient, err := os.Open(pathClient)
+		assert.Nil(t, err)
+
+		bufferClient := bufio.NewReader(fileClient)
+		tcpIDClient := &api.TcpID{
+			SrcIP:   "1",
+			DstIP:   "2",
+			SrcPort: "1",
+			DstPort: "2",
+		}
+		reqResMatcher := dissector.NewResponseRequestMatcher()
+		err = dissector.Dissect(bufferClient, true, tcpIDClient, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
+		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
+			panic(err)
+		}
+
+		// Response
+		pathServer := basePath + respSuffix
+		fmt.Printf("%s %s\n", msgDissecting, pathServer)
+		fileServer, err := os.Open(pathServer)
+		assert.Nil(t, err)
+
+		bufferServer := bufio.NewReader(fileServer)
+		tcpIDServer := &api.TcpID{
+			SrcIP:   "2",
+			DstIP:   "1",
+			SrcPort: "2",
+			DstPort: "1",
+		}
+		err = dissector.Dissect(bufferServer, false, tcpIDServer, counterPair, &api.SuperTimer{}, superIdentifier, emitter, options, reqResMatcher)
+		if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
+			panic(err)
+		}
+
+		fileClient.Close()
+		fileServer.Close()
+
+		pathExpect := path.Join(expectDirDissect, fmt.Sprintf("%s.json", basePath[4:]))
+
+		time.Sleep(10 * time.Millisecond)
+
+		stop <- true
+
+		marshaled, err := json.Marshal(items)
+		assert.Nil(t, err)
+
+		if testUpdateEnabled {
+			if len(items) > 0 {
+				err = os.WriteFile(pathExpect, marshaled, 0644)
+				assert.Nil(t, err)
+			}
+		} else {
+			if _, err := os.Stat(pathExpect); errors.Is(err, os.ErrNotExist) {
+				assert.Len(t, items, 0)
+			} else {
+				expectedBytes, err := ioutil.ReadFile(pathExpect)
+				assert.Nil(t, err)
+
+				assert.JSONEq(t, string(expectedBytes), string(marshaled))
+			}
+		}
+	}
+}
+
+func TestAnalyze(t *testing.T) {
+	_, testUpdateEnabled := os.LookupEnv(testUpdate)
+
+	expectDirDissect := path.Join(expectDir, dissectDir)
+	expectDirAnalyze := path.Join(expectDir, analyzeDir)
+
+	if testUpdateEnabled {
+		os.RemoveAll(expectDirAnalyze)
+		err := os.MkdirAll(expectDirAnalyze, 0775)
+		assert.Nil(t, err)
+	}
+
+	dissector := NewDissector()
+	paths, err := filepath.Glob(path.Join(expectDirDissect, patternDissect))
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	for _, _path := range paths {
+		fmt.Printf("%s %s\n", msgAnalyzing, _path)
+
+		bytes, err := ioutil.ReadFile(_path)
+		assert.Nil(t, err)
+
+		var items []*api.OutputChannelItem
+		err = json.Unmarshal(bytes, &items)
+		assert.Nil(t, err)
+
+		var entries []*api.Entry
+		for _, item := range items {
+			entry := dissector.Analyze(item, "", "", "")
+			entries = append(entries, entry)
+		}
+
+		pathExpect := path.Join(expectDirAnalyze, filepath.Base(_path))
+
+		marshaled, err := json.Marshal(entries)
+		assert.Nil(t, err)
+
+		if testUpdateEnabled {
+			if len(entries) > 0 {
+				err = os.WriteFile(pathExpect, marshaled, 0644)
+				assert.Nil(t, err)
+			}
+		} else {
+			if _, err := os.Stat(pathExpect); errors.Is(err, os.ErrNotExist) {
+				assert.Len(t, items, 0)
+			} else {
+				expectedBytes, err := ioutil.ReadFile(pathExpect)
+				assert.Nil(t, err)
+
+				assert.JSONEq(t, string(expectedBytes), string(marshaled))
+			}
+		}
+	}
+}
+
+func TestRepresent(t *testing.T) {
+	_, testUpdateEnabled := os.LookupEnv(testUpdate)
+
+	expectDirAnalyze := path.Join(expectDir, analyzeDir)
+	expectDirRepresent := path.Join(expectDir, representDir)
+
+	if testUpdateEnabled {
+		os.RemoveAll(expectDirRepresent)
+		err := os.MkdirAll(expectDirRepresent, 0775)
+		assert.Nil(t, err)
+	}
+
+	dissector := NewDissector()
+	paths, err := filepath.Glob(path.Join(expectDirAnalyze, patternDissect))
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	for _, _path := range paths {
+		fmt.Printf("%s %s\n", msgRepresenting, _path)
+
+		bytes, err := ioutil.ReadFile(_path)
+		assert.Nil(t, err)
+
+		var entries []*api.Entry
+		err = json.Unmarshal(bytes, &entries)
+		assert.Nil(t, err)
+
+		var objects []string
+		for _, entry := range entries {
+			object, _, err := dissector.Represent(entry.Request, entry.Response)
+			assert.Nil(t, err)
+			objects = append(objects, string(object))
+		}
+
+		pathExpect := path.Join(expectDirRepresent, filepath.Base(_path))
+
+		marshaled, err := json.Marshal(objects)
+		assert.Nil(t, err)
+
+		if testUpdateEnabled {
+			if len(objects) > 0 {
+				err = os.WriteFile(pathExpect, marshaled, 0644)
+				assert.Nil(t, err)
+			}
+		} else {
+			if _, err := os.Stat(pathExpect); errors.Is(err, os.ErrNotExist) {
+				assert.Len(t, objects, 0)
+			} else {
+				expectedBytes, err := ioutil.ReadFile(pathExpect)
+				assert.Nil(t, err)
+
+				assert.JSONEq(t, string(expectedBytes), string(marshaled))
+			}
+		}
+	}
+}

tap/extensions/amqp/read.go

@@ -140,7 +140,7 @@ func readTimestamp(r io.Reader) (v time.Time, err error) {
 	if err = binary.Read(r, binary.BigEndian, &sec); err != nil {
 		return
 	}
-	return time.Unix(sec, 0), nil
+	return time.Unix(sec, 0).UTC(), nil
 }
 
 /*

tap/extensions/amqp/types.go

@@ -223,41 +223,8 @@ type Decimal struct {
 //
 type Table map[string]interface{}
 
-func validateField(f interface{}) error {
-	switch fv := f.(type) {
-	case nil, bool, byte, int, int16, int32, int64, float32, float64, string, []byte, Decimal, time.Time:
-		return nil
-
-	case []interface{}:
-		for _, v := range fv {
-			if err := validateField(v); err != nil {
-				return fmt.Errorf("in array %s", err)
-			}
-		}
-		return nil
-
-	case Table:
-		for k, v := range fv {
-			if err := validateField(v); err != nil {
-				return fmt.Errorf("table field %q %s", k, err)
-			}
-		}
-		return nil
-	}
-
-	return fmt.Errorf("value %T not supported", f)
-}
-
-// Validate returns and error if any Go types in the table are incompatible with AMQP types.
-func (t Table) Validate() error {
-	return validateField(t)
-}
-
 type Message interface {
-	id() (uint16, uint16)
-	wait() bool
 	read(io.Reader) error
-	write(io.Writer) error
 }
 
 /*
@@ -286,8 +253,6 @@ system calls to read a frame.
 
 */
 type frame interface {
-	write(io.Writer) error
-	channel() uint16
 }
 
 type AmqpReader struct {
@@ -323,8 +288,6 @@ type MethodFrame struct {
 	Method    Message
 }
 
-func (f *MethodFrame) channel() uint16 { return f.ChannelId }
-
 /*
 Heartbeating is a technique designed to undo one of TCP/IP's features, namely
 its ability to recover from a broken physical connection by closing only after
@@ -338,8 +301,6 @@ type HeartbeatFrame struct {
 	ChannelId uint16
 }
 
-func (f *HeartbeatFrame) channel() uint16 { return f.ChannelId }
-
 /*
 Certain methods (such as Basic.Publish, Basic.Deliver, etc.) are formally
 defined as carrying content.  When a peer sends such a method frame, it always
@@ -367,8 +328,6 @@ type HeaderFrame struct {
 	Properties Properties
 }
 
-func (f *HeaderFrame) channel() uint16 { return f.ChannelId }
-
 /*
 Content is the application data we carry from client-to-client via the AMQP
 server.  Content is, roughly speaking, a set of properties plus a binary data
@@ -388,5 +347,3 @@ type BodyFrame struct {
 	ChannelId uint16
 	Body      []byte
 }
-
-func (f *BodyFrame) channel() uint16 { return f.ChannelId }

tap/extensions/amqp/write.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"
-	"math"
-	"time"
-)
-
-func (f *MethodFrame) write(w io.Writer) (err error) {
-	var payload bytes.Buffer
-
-	if f.Method == nil {
-		return errors.New("malformed frame: missing method")
-	}
-
-	class, method := f.Method.id()
-
-	if err = binary.Write(&payload, binary.BigEndian, class); err != nil {
-		return
-	}
-
-	if err = binary.Write(&payload, binary.BigEndian, method); err != nil {
-		return
-	}
-
-	if err = f.Method.write(&payload); err != nil {
-		return
-	}
-
-	return writeFrame(w, frameMethod, f.ChannelId, payload.Bytes())
-}
-
-// Heartbeat
-//
-// Payload is empty
-func (f *HeartbeatFrame) write(w io.Writer) (err error) {
-	return writeFrame(w, frameHeartbeat, f.ChannelId, []byte{})
-}
-
-// CONTENT HEADER
-// 0          2        4           12               14
-// +----------+--------+-----------+----------------+------------- - -
-// | class-id | weight | body size | property flags | property list...
-// +----------+--------+-----------+----------------+------------- - -
-//    short     short    long long       short        remainder...
-//
-func (f *HeaderFrame) write(w io.Writer) (err error) {
-	var payload bytes.Buffer
-	var zeroTime time.Time
-
-	if err = binary.Write(&payload, binary.BigEndian, f.ClassId); err != nil {
-		return
-	}
-
-	if err = binary.Write(&payload, binary.BigEndian, f.weight); err != nil {
-		return
-	}
-
-	if err = binary.Write(&payload, binary.BigEndian, f.Size); err != nil {
-		return
-	}
-
-	// First pass will build the mask to be serialized, second pass will serialize
-	// each of the fields that appear in the mask.
-
-	var mask uint16
-
-	if len(f.Properties.ContentType) > 0 {
-		mask = mask | flagContentType
-	}
-	if len(f.Properties.ContentEncoding) > 0 {
-		mask = mask | flagContentEncoding
-	}
-	if f.Properties.Headers != nil && len(f.Properties.Headers) > 0 {
-		mask = mask | flagHeaders
-	}
-	if f.Properties.DeliveryMode > 0 {
-		mask = mask | flagDeliveryMode
-	}
-	if f.Properties.Priority > 0 {
-		mask = mask | flagPriority
-	}
-	if len(f.Properties.CorrelationId) > 0 {
-		mask = mask | flagCorrelationId
-	}
-	if len(f.Properties.ReplyTo) > 0 {
-		mask = mask | flagReplyTo
-	}
-	if len(f.Properties.Expiration) > 0 {
-		mask = mask | flagExpiration
-	}
-	if len(f.Properties.MessageId) > 0 {
-		mask = mask | flagMessageId
-	}
-	if f.Properties.Timestamp != zeroTime {
-		mask = mask | flagTimestamp
-	}
-	if len(f.Properties.Type) > 0 {
-		mask = mask | flagType
-	}
-	if len(f.Properties.UserId) > 0 {
-		mask = mask | flagUserId
-	}
-	if len(f.Properties.AppId) > 0 {
-		mask = mask | flagAppId
-	}
-
-	if err = binary.Write(&payload, binary.BigEndian, mask); err != nil {
-		return
-	}
-
-	if hasProperty(mask, flagContentType) {
-		if err = writeShortstr(&payload, f.Properties.ContentType); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagContentEncoding) {
-		if err = writeShortstr(&payload, f.Properties.ContentEncoding); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagHeaders) {
-		if err = writeTable(&payload, f.Properties.Headers); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagDeliveryMode) {
-		if err = binary.Write(&payload, binary.BigEndian, f.Properties.DeliveryMode); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagPriority) {
-		if err = binary.Write(&payload, binary.BigEndian, f.Properties.Priority); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagCorrelationId) {
-		if err = writeShortstr(&payload, f.Properties.CorrelationId); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagReplyTo) {
-		if err = writeShortstr(&payload, f.Properties.ReplyTo); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagExpiration) {
-		if err = writeShortstr(&payload, f.Properties.Expiration); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagMessageId) {
-		if err = writeShortstr(&payload, f.Properties.MessageId); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagTimestamp) {
-		if err = binary.Write(&payload, binary.BigEndian, uint64(f.Properties.Timestamp.Unix())); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagType) {
-		if err = writeShortstr(&payload, f.Properties.Type); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagUserId) {
-		if err = writeShortstr(&payload, f.Properties.UserId); err != nil {
-			return
-		}
-	}
-	if hasProperty(mask, flagAppId) {
-		if err = writeShortstr(&payload, f.Properties.AppId); err != nil {
-			return
-		}
-	}
-
-	return writeFrame(w, frameHeader, f.ChannelId, payload.Bytes())
-}
-
-// Body
-//
-// Payload is one byterange from the full body who's size is declared in the
-// Header frame
-func (f *BodyFrame) write(w io.Writer) (err error) {
-	return writeFrame(w, frameBody, f.ChannelId, f.Body)
-}
-
-func writeFrame(w io.Writer, typ uint8, channel uint16, payload []byte) (err error) {
-	end := []byte{frameEnd}
-	size := uint(len(payload))
-
-	_, err = w.Write([]byte{
-		byte(typ),
-		byte((channel & 0xff00) >> 8),
-		byte((channel & 0x00ff) >> 0),
-		byte((size & 0xff000000) >> 24),
-		byte((size & 0x00ff0000) >> 16),
-		byte((size & 0x0000ff00) >> 8),
-		byte((size & 0x000000ff) >> 0),
-	})
-
-	if err != nil {
-		return
-	}
-
-	if _, err = w.Write(payload); err != nil {
-		return
-	}
-
-	if _, err = w.Write(end); err != nil {
-		return
-	}
-
-	return
-}
-
-func writeShortstr(w io.Writer, s string) (err error) {
-	b := []byte(s)
-
-	var length = uint8(len(b))
-
-	if err = binary.Write(w, binary.BigEndian, length); err != nil {
-		return
-	}
-
-	if _, err = w.Write(b[:length]); err != nil {
-		return
-	}
-
-	return
-}
-
-func writeLongstr(w io.Writer, s string) (err error) {
-	b := []byte(s)
-
-	var length = uint32(len(b))
-
-	if err = binary.Write(w, binary.BigEndian, length); err != nil {
-		return
-	}
-
-	if _, err = w.Write(b[:length]); err != nil {
-		return
-	}
-
-	return
-}
-
-/*
-'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 writeField(w io.Writer, value interface{}) (err error) {
-	var buf [9]byte
-	var enc []byte
-
-	switch v := value.(type) {
-	case bool:
-		buf[0] = 't'
-		if v {
-			buf[1] = byte(1)
-		} else {
-			buf[1] = byte(0)
-		}
-		enc = buf[:2]
-
-	case byte:
-		buf[0] = 'b'
-		buf[1] = byte(v)
-		enc = buf[:2]
-
-	case int16:
-		buf[0] = 's'
-		binary.BigEndian.PutUint16(buf[1:3], uint16(v))
-		enc = buf[:3]
-
-	case int:
-		buf[0] = 'I'
-		binary.BigEndian.PutUint32(buf[1:5], uint32(v))
-		enc = buf[:5]
-
-	case int32:
-		buf[0] = 'I'
-		binary.BigEndian.PutUint32(buf[1:5], uint32(v))
-		enc = buf[:5]
-
-	case int64:
-		buf[0] = 'l'
-		binary.BigEndian.PutUint64(buf[1:9], uint64(v))
-		enc = buf[:9]
-
-	case float32:
-		buf[0] = 'f'
-		binary.BigEndian.PutUint32(buf[1:5], math.Float32bits(v))
-		enc = buf[:5]
-
-	case float64:
-		buf[0] = 'd'
-		binary.BigEndian.PutUint64(buf[1:9], math.Float64bits(v))
-		enc = buf[:9]
-
-	case Decimal:
-		buf[0] = 'D'
-		buf[1] = byte(v.Scale)
-		binary.BigEndian.PutUint32(buf[2:6], uint32(v.Value))
-		enc = buf[:6]
-
-	case string:
-		buf[0] = 'S'
-		binary.BigEndian.PutUint32(buf[1:5], uint32(len(v)))
-		enc = append(buf[:5], []byte(v)...)
-
-	case []interface{}: // field-array
-		buf[0] = 'A'
-
-		sec := new(bytes.Buffer)
-		for _, val := range v {
-			if err = writeField(sec, val); err != nil {
-				return
-			}
-		}
-
-		binary.BigEndian.PutUint32(buf[1:5], uint32(sec.Len()))
-		if _, err = w.Write(buf[:5]); err != nil {
-			return
-		}
-
-		if _, err = w.Write(sec.Bytes()); err != nil {
-			return
-		}
-
-		return
-
-	case time.Time:
-		buf[0] = 'T'
-		binary.BigEndian.PutUint64(buf[1:9], uint64(v.Unix()))
-		enc = buf[:9]
-
-	case Table:
-		if _, err = w.Write([]byte{'F'}); err != nil {
-			return
-		}
-		return writeTable(w, v)
-
-	case []byte:
-		buf[0] = 'x'
-		binary.BigEndian.PutUint32(buf[1:5], uint32(len(v)))
-		if _, err = w.Write(buf[0:5]); err != nil {
-			return
-		}
-		if _, err = w.Write(v); err != nil {
-			return
-		}
-		return
-
-	case nil:
-		buf[0] = 'V'
-		enc = buf[:1]
-
-	default:
-		return ErrFieldType
-	}
-
-	_, err = w.Write(enc)
-
-	return
-}
-
-func writeTable(w io.Writer, table Table) (err error) {
-	var buf bytes.Buffer
-
-	for key, val := range table {
-		if err = writeShortstr(&buf, key); err != nil {
-			return
-		}
-		if err = writeField(&buf, val); err != nil {
-			return
-		}
-	}
-
-	return writeLongstr(w, buf.String())
-}