-
Notifications
You must be signed in to change notification settings - Fork 789
/
json.go
63 lines (53 loc) · 1.66 KB
/
json.go
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
package io
import (
"encoding/json"
"github.com/Jeffail/benthos/v3/lib/message"
"github.com/Jeffail/benthos/v3/lib/message/metadata"
"github.com/Jeffail/benthos/v3/lib/types"
)
//------------------------------------------------------------------------------
type partJSONStruct struct {
Metadata map[string]string `json:"metadata"`
Value string `json:"value"`
}
func partToJSONStruct(p types.Part) partJSONStruct {
meta := map[string]string{}
p.Metadata().Iter(func(k, v string) error {
meta[k] = v
return nil
})
return partJSONStruct{
Metadata: meta,
Value: string(p.Get()),
}
}
func partFromJSONStruct(p partJSONStruct) types.Part {
part := message.NewPart([]byte(p.Value))
part.SetMetadata(metadata.New(p.Metadata))
return part
}
//------------------------------------------------------------------------------
// MessageToJSON converts a message into raw JSON bytes of the form:
// [{"value":"foo","metadata":{"bar":"baz"}}]
func MessageToJSON(msg types.Message) ([]byte, error) {
message := []partJSONStruct{}
msg.Iter(func(i int, part types.Part) error {
message = append(message, partToJSONStruct(part))
return nil
})
return json.Marshal(message)
}
// MessageFromJSON parses JSON bytes into a message type of the form:
// [{"value":"foo","metadata":{"bar":"baz"}}]
func MessageFromJSON(jsonBytes []byte) (types.Message, error) {
var jsonParts []partJSONStruct
if err := json.Unmarshal(jsonBytes, &jsonParts); err != nil {
return nil, err
}
msg := message.New(nil)
for _, v := range jsonParts {
msg.Append(partFromJSONStruct(v))
}
return msg, nil
}
//------------------------------------------------------------------------------