flow/message.go at master · bytepowered/flow · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
package flow

import (
	"bytes"
	"time"
)

var _ Message = NewObjectMessage(Header{}, nil)

type ObjectMessage struct {
	Headers   Header
	data      interface{}
	timestamp time.Time
}

func NewObjectMessage(header Header, data interface{}) *ObjectMessage {
	return &ObjectMessage{
		Headers:   header,
		data:      data,
		timestamp: time.UnixMilli(header.Unix),
	}
}

func (m *ObjectMessage) ID() int64 {
	return m.Headers.Id
}

func (m *ObjectMessage) Tag() string {
	return m.Headers.Tag
}

func (m *ObjectMessage) Kind() Kind {
	return m.Headers.Kind
}

func (m *ObjectMessage) Time() time.Time {
	return m.timestamp
}

func (m *ObjectMessage) Header() Header {
	return m.Headers
}

func (m *ObjectMessage) Data() interface{} {
	return m.data
}

//// Bytes

var _ Message = NewFrameMessage(Header{}, nil)

type FrameMessage struct {
	*ObjectMessage
}

func NewFrameMessage(header Header, frame []byte) *FrameMessage {
	return &FrameMessage{
		ObjectMessage: NewObjectMessage(header, frame),
	}
}

func (e *FrameMessage) Bytes() []byte {
	return e.Data().([]byte)
}

//// Text

var _ Message = NewTextMessage(Header{}, "")

type TextFrame struct {
	*ObjectMessage
}

func NewTextMessage(header Header, text string) *TextFrame {
	return &TextFrame{
		ObjectMessage: NewObjectMessage(header, text),
	}
}

func (e *TextFrame) Text() string {
	return e.Data().(string)
}

//// Fields

var _ Message = NewFieldsMessage(Header{}, nil)

type FieldsMessage struct {
	*ObjectMessage
}

func NewFieldsMessage(header Header, fields []string) *FieldsMessage {
	return &FieldsMessage{
		ObjectMessage: NewObjectMessage(header, fields),
	}
}

func (e *FieldsMessage) Fields() []string {
	return e.Data().([]string)
}

//// Buffer

var _ Message = NewBufferMessage(Header{}, nil)

type BufferMessage struct {
	*ObjectMessage
}

func NewBufferMessage(header Header, buffer *bytes.Buffer) *BufferMessage {
	return &BufferMessage{
		ObjectMessage: NewObjectMessage(header, buffer),
	}
}

func (e *BufferMessage) Buffer() *bytes.Buffer {
	return e.Data().(*bytes.Buffer)
}