OrgGo/Mainflux.mainflux
1
0
Fork 0
mirror of https://github.com/mainflux/mainflux.git synced 2025-04-26 13:48:53 +08:00
Code Issues Projects Releases Wiki Activity

MF-1088 - Remove message payload content type (#1121)

Browse Source 
* MF-1088 - Remove message payload content type

Signed-off-by: Manuel Imperiale <manuel.imperiale@gmail.com>

* Add default contentFormat

Signed-off-by: Manuel Imperiale <manuel.imperiale@gmail.com>
This commit is contained in:
Manuel Imperiale 2020-04-21 19:33:11 +02:00 committed by GitHub
parent ccbfe20607
commit 7a18d3fa56
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 196 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.env
View File

@ -125,6 +125,7 @@ MF_CASSANDRA_WRITER_PORT=8902
MF_CASSANDRA_WRITER_DB_PORT=9042
MF_CASSANDRA_WRITER_DB_CLUSTER=mainflux-cassandra
MF_CASSANDRA_WRITER_DB_KEYSPACE=mainflux
MF_CASSANDRA_WRITER_CONTENT_TYPE=application/senml+json
### Cassandra Reader
MF_CASSANDRA_READER_LOG_LEVEL=debug
@ -145,6 +146,7 @@ MF_INFLUX_WRITER_DB_USER=mainflux
MF_INFLUX_WRITER_DB_PASS=mainflux
MF_INFLUX_WRITER_DB=mainflux
MF_INFLUX_WRITER_GRAFANA_PORT=3001
MF_INFLUX_WRITER_CONTENT_TYPE=application/senml+json
### InfluxDB Reader
MF_INFLUX_READER_LOG_LEVEL=debug
@ -161,6 +163,7 @@ MF_MONGO_WRITER_LOG_LEVEL=debug
MF_MONGO_WRITER_PORT=8901
MF_MONGO_WRITER_DB=mainflux
MF_MONGO_WRITER_DB_PORT=27017
MF_MONGO_WRITER_CONTENT_TYPE=application/senml+json
### MongoDB Reader
MF_MONGO_READER_LOG_LEVEL=debug
@ -181,6 +184,7 @@ MF_POSTGRES_WRITER_DB_SSL_MODE=disable
MF_POSTGRES_WRITER_DB_SSL_CERT=""
MF_POSTGRES_WRITER_DB_SSL_KEY=""
MF_POSTGRES_WRITER_DB_SSL_ROOT_CERT=""
MF_POSTGRES_WRITER_CONTENT_TYPE=application/senml+json
### Postgres Reader
MF_POSTGRES_READER_LOG_LEVEL=debug

41
broker/message.pb.go
View File

@ -27,9 +27,8 @@ type Message struct {
Subtopic string `protobuf:"bytes,2,opt,name=subtopic,proto3" json:"subtopic,omitempty"`
Publisher string `protobuf:"bytes,3,opt,name=publisher,proto3" json:"publisher,omitempty"`
Protocol string `protobuf:"bytes,4,opt,name=protocol,proto3" json:"protocol,omitempty"`
ContentType string `protobuf:"bytes,5,opt,name=contentType,proto3" json:"contentType,omitempty"`
Payload []byte `protobuf:"bytes,6,opt,name=payload,proto3" json:"payload,omitempty"`
Created *timestamp.Timestamp `protobuf:"bytes,7,opt,name=created,proto3" json:"created,omitempty"`
Payload []byte `protobuf:"bytes,5,opt,name=payload,proto3" json:"payload,omitempty"`
Created *timestamp.Timestamp `protobuf:"bytes,6,opt,name=created,proto3" json:"created,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
@ -87,13 +86,6 @@ func (m *Message) GetProtocol() string {
return ""
}
func (m *Message) GetContentType() string {
if m != nil {
return m.ContentType
}
return ""
}
func (m *Message) GetPayload() []byte {
if m != nil {
return m.Payload
@ -115,19 +107,18 @@ func init() {
func init() { proto.RegisterFile("broker/message.proto", fileDescriptor_6357da820a7eacc2) }
var fileDescriptor_6357da820a7eacc2 = []byte{
// 218 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x4c, 0x8f, 0xcd, 0x4e, 0xc4, 0x20,
0x14, 0x85, 0x83, 0x3f, 0xad, 0xc3, 0xb8, 0x22, 0x2e, 0x6e, 0x1a, 0x13, 0x1b, 0x57, 0x5d, 0xd1,
0x44, 0x7d, 0x0d, 0x37, 0xcd, 0xbc, 0x00, 0x30, 0xd7, 0x4e, 0x23, 0xe5, 0x12, 0xa0, 0x8b, 0x79,
0x66, 0x5f, 0xc2, 0x0c, 0x04, 0x9d, 0xe5, 0xf9, 0xbe, 0x7b, 0x02, 0x87, 0x3f, 0xe9, 0x40, 0xdf,
0x18, 0xc6, 0x15, 0x63, 0x54, 0x33, 0x4a, 0x1f, 0x28, 0x91, 0x68, 0x0a, 0xed, 0x5e, 0x66, 0xa2,
0xd9, 0xe2, 0x98, 0xa9, 0xde, 0xbe, 0xc6, 0xb4, 0xac, 0x18, 0x93, 0x5a, 0x7d, 0x39, 0x7c, 0xfd,
0x61, 0xbc, 0xfd, 0x2c, 0x55, 0x01, 0xbc, 0x35, 0x27, 0xe5, 0x1c, 0x5a, 0x60, 0x3d, 0x1b, 0x76,
0x53, 0x8d, 0xa2, 0xe3, 0x0f, 0x71, 0xd3, 0x89, 0xfc, 0x62, 0xe0, 0x26, 0xab, 0xbf, 0x2c, 0x9e,
0xf9, 0xce, 0x6f, 0xda, 0x2e, 0xf1, 0x84, 0x01, 0x6e, 0xb3, 0xfc, 0x07, 0x97, 0x66, 0x7e, 0xc8,
0x90, 0x85, 0xbb, 0xd2, 0xac, 0x59, 0xf4, 0x7c, 0x6f, 0xc8, 0x25, 0x74, 0xe9, 0x70, 0xf6, 0x08,
0xf7, 0x59, 0x5f, 0xa3, 0xcb, 0x8f, 0xbc, 0x3a, 0x5b, 0x52, 0x47, 0x68, 0x7a, 0x36, 0x3c, 0x4e,
0x35, 0x8a, 0x0f, 0xde, 0x9a, 0x80, 0x2a, 0xe1, 0x11, 0xda, 0x9e, 0x0d, 0xfb, 0xb7, 0x4e, 0x96,
0xa9, 0xb2, 0x4e, 0x95, 0x87, 0x3a, 0x75, 0xaa, 0xa7, 0xba, 0xc9, 0xf2, 0xfd, 0x37, 0x00, 0x00,
0xff, 0xff, 0x39, 0x63, 0xf9, 0x79, 0x35, 0x01, 0x00, 0x00,
// 202 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x44, 0x8f, 0xcb, 0x6e, 0x84, 0x20,
0x18, 0x85, 0x43, 0x2f, 0x5a, 0x69, 0x57, 0xa4, 0x0b, 0x62, 0x9a, 0xd4, 0x74, 0xe5, 0x0a, 0x93,
0xb6, 0xaf, 0x31, 0x1b, 0x33, 0x2f, 0x00, 0xf8, 0x8f, 0x9a, 0x41, 0x7f, 0x02, 0xb8, 0x98, 0x47,
0x9c, 0xb7, 0x9a, 0x08, 0x61, 0x66, 0xf9, 0x9d, 0x4b, 0x4e, 0x0e, 0xfd, 0x54, 0x0e, 0xcf, 0xe0,
0xba, 0x05, 0xbc, 0x97, 0x23, 0x08, 0xeb, 0x30, 0x20, 0x2b, 0x92, 0x5a, 0x7f, 0x8f, 0x88, 0xa3,
0x81, 0x2e, 0xaa, 0x6a, 0x3b, 0x75, 0x61, 0x5e, 0xc0, 0x07, 0xb9, 0xd8, 0x14, 0xfc, 0xb9, 0x12,
0x5a, 0x1e, 0x52, 0x95, 0x71, 0x5a, 0xea, 0x49, 0xae, 0x2b, 0x18, 0x4e, 0x1a, 0xd2, 0x56, 0x7d,
0x46, 0x56, 0xd3, 0x37, 0xbf, 0xa9, 0x80, 0x76, 0xd6, 0xfc, 0x29, 0x5a, 0x77, 0x66, 0x5f, 0xb4,
0xb2, 0x9b, 0x32, 0xb3, 0x9f, 0xc0, 0xf1, 0xe7, 0x68, 0x3e, 0x84, 0xbd, 0x19, 0x87, 0x34, 0x1a,
0xfe, 0x92, 0x9a, 0x99, 0xf7, 0x3d, 0x2b, 0x2f, 0x06, 0xe5, 0xc0, 0x5f, 0x1b, 0xd2, 0x7e, 0xf4,
0x19, 0xd9, 0x3f, 0x2d, 0xb5, 0x03, 0x19, 0x60, 0xe0, 0x45, 0x43, 0xda, 0xf7, 0xdf, 0x5a, 0xa4,
0x23, 0x22, 0x1f, 0x11, 0xc7, 0x7c, 0xa4, 0xcf, 0x51, 0x55, 0x44, 0xf3, 0xef, 0x16, 0x00, 0x00,
0xff, 0xff, 0x0b, 0xd8, 0x04, 0x71, 0x13, 0x01, 0x00, 0x00,
}

5
broker/message.proto
View File

@ -11,7 +11,6 @@ message Message {
string subtopic = 2;
string publisher = 3;
string protocol = 4;
string contentType = 5;
bytes payload = 6;
google.protobuf.Timestamp created = 7;
bytes payload = 5;
google.protobuf.Timestamp created = 6;
}

10
cmd/cassandra-writer/main.go
View File

@ -38,6 +38,7 @@ const (
defDBPass = "mainflux"
defDBPort = "9042"
defSubjectsCfgPath = "/config/subjects.toml"
defContentType = "application/senml+json"
envNatsURL = "MF_NATS_URL"
envLogLevel = "MF_CASSANDRA_WRITER_LOG_LEVEL"
@ -48,14 +49,16 @@ const (
envDBPass = "MF_CASSANDRA_WRITER_DB_PASS"
envDBPort = "MF_CASSANDRA_WRITER_DB_PORT"
envSubjectsCfgPath = "MF_CASSANDRA_WRITER_SUBJECTS_CONFIG"
envContentType = "MF_CASSANDRA_WRITER_CONTENT_TYPE"
)
type config struct {
natsURL string
logLevel string
port string
dbCfg cassandra.DBConfig
subjectsCfgPath string
contentType string
dbCfg cassandra.DBConfig
}
func main() {
@ -77,7 +80,7 @@ func main() {
defer session.Close()
repo := newService(session, logger)
st := senml.New()
st := senml.New(cfg.contentType)
if err := writers.Start(b, repo, st, svcName, cfg.subjectsCfgPath, logger); err != nil {
logger.Error(fmt.Sprintf("Failed to create Cassandra writer: %s", err))
}
@ -114,8 +117,9 @@ func loadConfig() config {
natsURL: mainflux.Env(envNatsURL, defNatsURL),
logLevel: mainflux.Env(envLogLevel, defLogLevel),
port: mainflux.Env(envPort, defPort),
dbCfg: dbCfg,
subjectsCfgPath: mainflux.Env(envSubjectsCfgPath, defSubjectsCfgPath),
contentType: mainflux.Env(envContentType, defContentType),
dbCfg: dbCfg,
}
}

6
cmd/influxdb-writer/main.go
View File

@ -35,6 +35,7 @@ const (
defDBUser = "mainflux"
defDBPass = "mainflux"
defSubjectsCfgPath = "/config/subjects.toml"
defContentType = "application/senml+json"
envNatsURL = "MF_NATS_URL"
envLogLevel = "MF_INFLUX_WRITER_LOG_LEVEL"
@ -45,6 +46,7 @@ const (
envDBUser = "MF_INFLUX_WRITER_DB_USER"
envDBPass = "MF_INFLUX_WRITER_DB_PASS"
envSubjectsCfgPath = "MF_INFLUX_WRITER_SUBJECTS_CONFIG"
envContentType = "MF_INFLUX_WRITER_CONTENT_TYPE"
)
type config struct {
@ -57,6 +59,7 @@ type config struct {
dbUser string
dbPass string
subjectsCfgPath string
contentType string
}
func main() {
@ -86,7 +89,7 @@ func main() {
counter, latency := makeMetrics()
repo = api.LoggingMiddleware(repo, logger)
repo = api.MetricsMiddleware(repo, counter, latency)
st := senml.New()
st := senml.New(cfg.contentType)
if err := writers.Start(b, repo, st, svcName, cfg.subjectsCfgPath, logger); err != nil {
logger.Error(fmt.Sprintf("Failed to start InfluxDB writer: %s", err))
os.Exit(1)
@ -116,6 +119,7 @@ func loadConfigs() (config, influxdata.HTTPConfig) {
dbUser: mainflux.Env(envDBUser, defDBUser),
dbPass: mainflux.Env(envDBPass, defDBPass),
subjectsCfgPath: mainflux.Env(envSubjectsCfgPath, defSubjectsCfgPath),
contentType: mainflux.Env(envContentType, defContentType),
}
clientCfg := influxdata.HTTPConfig{

6
cmd/mongodb-writer/main.go
View File

@ -35,6 +35,7 @@ const (
defDBHost = "localhost"
defDBPort = "27017"
defSubjectsCfgPath = "/config/subjects.toml"
defContentType = "application/senml+json"
envNatsURL = "MF_NATS_URL"
envLogLevel = "MF_MONGO_WRITER_LOG_LEVEL"
@ -43,6 +44,7 @@ const (
envDBHost = "MF_MONGO_WRITER_DB_HOST"
envDBPort = "MF_MONGO_WRITER_DB_PORT"
envSubjectsCfgPath = "MF_MONGO_WRITER_SUBJECTS_CONFIG"
envContentType = "MF_MONGO_WRITER_CONTENT_TYPE"
)
type config struct {
@ -53,6 +55,7 @@ type config struct {
dbHost string
dbPort string
subjectsCfgPath string
contentType string
}
func main() {
@ -83,7 +86,7 @@ func main() {
counter, latency := makeMetrics()
repo = api.LoggingMiddleware(repo, logger)
repo = api.MetricsMiddleware(repo, counter, latency)
st := senml.New()
st := senml.New(cfg.contentType)
if err := writers.Start(b, repo, st, svcName, cfg.subjectsCfgPath, logger); err != nil {
logger.Error(fmt.Sprintf("Failed to start MongoDB writer: %s", err))
os.Exit(1)
@ -111,6 +114,7 @@ func loadConfigs() config {
dbHost: mainflux.Env(envDBHost, defDBHost),
dbPort: mainflux.Env(envDBPort, defDBPort),
subjectsCfgPath: mainflux.Env(envSubjectsCfgPath, defSubjectsCfgPath),
contentType: mainflux.Env(envContentType, defContentType),
}
}

10
cmd/postgres-writer/main.go
View File

@ -40,6 +40,7 @@ const (
defDBSSLKey = ""
defDBSSLRootCert = ""
defSubjectsCfgPath = "/config/subjects.toml"
defContentType = "application/senml+json"
envNatsURL = "MF_NATS_URL"
envLogLevel = "MF_POSTGRES_WRITER_LOG_LEVEL"
@ -54,14 +55,16 @@ const (
envDBSSLKey = "MF_POSTGRES_WRITER_DB_SSL_KEY"
envDBSSLRootCert = "MF_POSTGRES_WRITER_DB_SSL_ROOT_CERT"
envSubjectsCfgPath = "MF_POSTGRES_WRITER_SUBJECTS_CONFIG"
envContentType = "MF_POSTGRES_WRITER_CONTENT_TYPE"
)
type config struct {
natsURL string
logLevel string
port string
dbConfig postgres.Config
subjectsCfgPath string
contentType string
dbConfig postgres.Config
}
func main() {
@ -83,7 +86,7 @@ func main() {
defer db.Close()
repo := newService(db, logger)
st := senml.New()
st := senml.New(cfg.contentType)
if err = writers.Start(b, repo, st, svcName, cfg.subjectsCfgPath, logger); err != nil {
logger.Error(fmt.Sprintf("Failed to create Postgres writer: %s", err))
}
@ -119,8 +122,9 @@ func loadConfig() config {
natsURL: mainflux.Env(envNatsURL, defNatsURL),
logLevel: mainflux.Env(envLogLevel, defLogLevel),
port: mainflux.Env(envPort, defPort),
dbConfig: dbConfig,
subjectsCfgPath: mainflux.Env(envSubjectsCfgPath, defSubjectsCfgPath),
contentType: mainflux.Env(envContentType, defContentType),
dbConfig: dbConfig,
}
}

45
coap/api/transport.go
View File

@ -23,7 +23,6 @@ import (
"github.com/mainflux/mainflux/broker"
"github.com/mainflux/mainflux/coap"
log "github.com/mainflux/mainflux/logger"
"github.com/mainflux/mainflux/transformers/senml"
"github.com/prometheus/client_golang/prometheus/promhttp"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
@ -220,11 +219,6 @@ func receive(svc coap.Service, msg *gocoap.Message) *gocoap.Message {
return res
}
ct, err := contentType(msg)
if err != nil {
ct = ""
}
publisher, err := authorize(msg, res, chanID)
if err != nil {
res.Code = gocoap.Forbidden
@ -237,13 +231,12 @@ func receive(svc coap.Service, msg *gocoap.Message) *gocoap.Message {
}
m := broker.Message{
Channel: chanID,
Subtopic: subtopic,
Publisher: publisher,
ContentType: ct,
Protocol: protocol,
Payload: msg.Payload,
Created: created,
Channel: chanID,
Subtopic: subtopic,
Publisher: publisher,
Protocol: protocol,
Payload: msg.Payload,
Created: created,
}
if err := svc.Publish(context.Background(), "", m); err != nil {
@ -341,15 +334,6 @@ func handleMessage(conn *net.UDPConn, addr *net.UDPAddr, o *coap.Observer, msg *
observeVal := buff.Bytes()
notifyMsg.SetOption(gocoap.Observe, observeVal[len(observeVal)-3:])
coapCT := senMLJSON
switch msg.ContentType {
case senml.JSON:
coapCT = senMLJSON
case senml.CBOR:
coapCT = senMLCBOR
}
notifyMsg.SetOption(gocoap.ContentFormat, coapCT)
if err := gocoap.Transmit(conn, addr, notifyMsg); err != nil {
logger.Warn(fmt.Sprintf("Failed to send message to observer: %s", err))
}
@ -394,20 +378,3 @@ func ping(svc coap.Service, obsID string, conn *net.UDPConn, addr *net.UDPAddr,
}
}
}
func contentType(msg *gocoap.Message) (string, error) {
ctid, ok := msg.Option(gocoap.ContentFormat).(gocoap.MediaType)
if !ok {
return "", errBadRequest
}
ct := ""
switch ctid {
case senMLJSON:
ct = senml.JSON
case senMLCBOR:
ct = senml.CBOR
}
return ct, nil
}

12
http/api/transport.go
View File

@ -115,14 +115,12 @@ func decodeRequest(ctx context.Context, r *http.Request) (interface{}, error) {
return nil, err
}
ct := r.Header.Get("Content-Type")
msg := broker.Message{
Protocol: protocol,
ContentType: ct,
Channel: chanID,
Subtopic: subtopic,
Payload: payload,
Created: created,
Protocol: protocol,
Channel: chanID,
Subtopic: subtopic,
Payload: payload,
Created: created,
}
req := publishReq{

12
lora/adapter.go
View File

@ -15,7 +15,6 @@ const (
protocol = "lora"
thingSuffix = "thing"
channelSuffix = "channel"
contentType = "application/json"
)
var (
@ -109,12 +108,11 @@ func (as *adapterService) Publish(ctx context.Context, token string, m Message)
// Publish on Mainflux NATS broker
msg := broker.Message{
Publisher: thing,
Protocol: protocol,
ContentType: contentType,
Channel: channel,
Payload: payload,
Created: created,
Publisher: thing,
Protocol: protocol,
Channel: channel,
Payload: payload,
Created: created,
}
return as.broker.Publish(ctx, token, msg)

20
mqtt/adapter.go
View File

@ -27,7 +27,6 @@ const protocol = "mqtt"
var (
channelRegExp = regexp.MustCompile(`^\/?channels\/([\w\-]+)\/messages(\/[^?]*)?(\?.*)?$`)
ctRegExp = regexp.MustCompile(`^(\/.*)?\/ct\/([^\/]+)$`)
errMalformedTopic = errors.New("malformed topic")
errMalformedData = errors.New("malformed request data")
errMalformedSubtopic = errors.New("malformed subtopic")
@ -149,12 +148,6 @@ func (e *Event) Publish(c *session.Client, topic *string, payload *[]byte) {
chanID := channelParts[1]
subtopic := channelParts[2]
ct := ""
if stParts := ctRegExp.FindStringSubmatch(subtopic); len(stParts) > 1 {
ct = stParts[2]
subtopic = stParts[1]
}
subtopic, err := parseSubtopic(subtopic)
if err != nil {
e.logger.Info("Error in mqtt publish: " + err.Error())
@ -168,13 +161,12 @@ func (e *Event) Publish(c *session.Client, topic *string, payload *[]byte) {
}
msg := broker.Message{
Protocol: protocol,
ContentType: ct,
Channel: chanID,
Subtopic: subtopic,
Publisher: c.Username,
Payload: *payload,
Created: created,
Protocol: protocol,
Channel: chanID,
Subtopic: subtopic,
Publisher: c.Username,
Payload: *payload,
Created: created,
}
if err := e.broker.Publish(context.TODO(), "", msg); err != nil {

13
opcua/gopcua/subscribe.go
View File

@ -234,13 +234,12 @@ func (c client) publish(token string, m message) error {
SenML := fmt.Sprintf(`[{"n":"%s", "t": %d, "%s":%v}]`, m.Type, m.Time, m.DataKey, m.Data)
payload := []byte(SenML)
msg := broker.Message{
Publisher: thingID,
Protocol: protocol,
ContentType: "Content-Type",
Channel: chanID,
Payload: payload,
Subtopic: m.NodeID,
Created: created,
Publisher: thingID,
Protocol: protocol,
Channel: chanID,
Payload: payload,
Subtopic: m.NodeID,
Created: created,
}
if err := c.broker.Publish(c.ctx, token, msg); err != nil {

8
transformers/senml/message.go
View File

@ -1,13 +1,5 @@
package senml
const (
// JSON represents SenML in JSON format content type.
JSON = "application/senml+json"
// CBOR represents SenML in CBOR format content type.
CBOR = "application/senml+cbor"
)
// Message represents a resolved (normalized) SenML record.
type Message struct {
Channel string `json:"channel,omitempty"`

29
transformers/senml/transformer.go
View File

@ -10,6 +10,13 @@ import (
"github.com/mainflux/senml"
)
const (
// JSON represents SenML in JSON format content type.
JSON = "application/senml+json"
// CBOR represents SenML in CBOR format content type.
CBOR = "application/senml+cbor"
)
var (
errDecode = errors.New("failed to decode senml")
errNormalize = errors.New("faled to normalize senml")
@ -20,20 +27,24 @@ var formats = map[string]senml.Format{
CBOR: senml.CBOR,
}
type transformer struct{}
// New returns transformer service implementation for SenML messages.
func New() transformers.Transformer {
return transformer{}
type transformer struct {
format senml.Format
}
func (n transformer) Transform(msg broker.Message) (interface{}, error) {
format, ok := formats[msg.ContentType]
// New returns transformer service implementation for SenML messages.
func New(contentFormat string) transformers.Transformer {
format, ok := formats[contentFormat]
if !ok {
format = senml.JSON
format = formats[JSON]
}
raw, err := senml.Decode(msg.Payload, format)
return transformer{
format: format,
}
}
func (t transformer) Transform(msg broker.Message) (interface{}, error) {
raw, err := senml.Decode(msg.Payload, t.format)
if err != nil {
return nil, errors.Wrap(errDecode, err)
}

91
transformers/senml/transformer_test.go
View File

@ -16,20 +16,14 @@ import (
"github.com/stretchr/testify/require"
)
func TestTransform(t *testing.T) {
func TestTransformJSON(t *testing.T) {
// Following hex-encoded bytes correspond to the content of:
// [{-2: "base-name", -3: 100.0, -4: "base-unit", -1: 10, -5: 10.0, -6: 100.0, 0: "name", 1: "unit", 6: 300.0, 7: 150.0, 2: 42.0, 5: 10.0}]
// For more details for mapping SenML labels to integers, please take a look here: https://tools.ietf.org/html/rfc8428#page-19.
cborBytes, err := hex.DecodeString("81ac2169626173652d6e616d6522fb40590000000000002369626173652d756e6974200a24fb402400000000000025fb405900000000000000646e616d650164756e697406fb4072c0000000000007fb4062c0000000000002fb404500000000000005fb4024000000000000")
require.Nil(t, err, "Decoding CBOR expected to succeed")
jsonBytes, err := hex.DecodeString("5b7b22626e223a22626173652d6e616d65222c226274223a3130302c226275223a22626173652d756e6974222c2262766572223a31302c226276223a31302c226273223a3130302c226e223a226e616d65222c2275223a22756e6974222c2274223a3330302c227574223a3135302c2276223a34322c2273223a31307d5d")
require.Nil(t, err, "Decoding CBOR expected to succeed")
require.Nil(t, err, "Decoding JSON expected to succeed")
tooManyBytes, err := hex.DecodeString("82AD2169626173652D6E616D6522F956402369626173652D756E6974200A24F9490025F9564000646E616D650164756E697406F95CB0036331323307F958B002F9514005F94900AA2169626173652D6E616D6522F956402369626173652D756E6974200A24F9490025F9564000646E616D6506F95CB007F958B005F94900")
require.Nil(t, err, "Decoding CBOR expected to succeed")
tr := senml.New()
tr := senml.New(senml.JSON)
msg := broker.Message{
Channel: "channel",
Subtopic: "subtopic",
@ -41,15 +35,76 @@ func TestTransform(t *testing.T) {
// 82AD2169626173652D6E616D6522F956402369626173652D756E6974200A24F9490025F9564000646E616D650164756E697406F95CB0036331323307F958B002F9514005F94900AA2169626173652D6E616D6522F956402369626173652D756E6974200A24F9490025F9564000646E616D6506F95CB007F958B005F94900
jsonPld := msg
jsonPld.ContentType = senml.JSON
jsonPld.Payload = jsonBytes
val := 52.0
sum := 110.0
msgs := []senml.Message{senml.Message{
Channel: "channel",
Subtopic: "subtopic",
Publisher: "publisher",
Protocol: "protocol",
Name: "base-namename",
Unit: "unit",
Time: 400,
UpdateTime: 150,
Value: &val,
Sum: &sum,
},
}
cases := []struct {
desc string
msg broker.Message
msgs interface{}
err error
}{
{
desc: "test normalize JSON",
msg: jsonPld,
msgs: msgs,
err: nil,
},
{
desc: "test normalize defaults to JSON",
msg: msg,
msgs: msgs,
err: nil,
},
}
for _, tc := range cases {
msgs, err := tr.Transform(tc.msg)
assert.Equal(t, tc.msgs, msgs, fmt.Sprintf("%s expected %v, got %v", tc.desc, tc.msgs, msgs))
assert.True(t, errors.Contains(err, tc.err), fmt.Sprintf("%s expected %s, got %s", tc.desc, tc.err, err))
}
}
func TestTransformCBOR(t *testing.T) {
// Following hex-encoded bytes correspond to the content of:
// [{-2: "base-name", -3: 100.0, -4: "base-unit", -1: 10, -5: 10.0, -6: 100.0, 0: "name", 1: "unit", 6: 300.0, 7: 150.0, 2: 42.0, 5: 10.0}]
// For more details for mapping SenML labels to integers, please take a look here: https://tools.ietf.org/html/rfc8428#page-19.
cborBytes, err := hex.DecodeString("81ac2169626173652d6e616d6522fb40590000000000002369626173652d756e6974200a24fb402400000000000025fb405900000000000000646e616d650164756e697406fb4072c0000000000007fb4062c0000000000002fb404500000000000005fb4024000000000000")
require.Nil(t, err, "Decoding CBOR expected to succeed")
tooManyBytes, err := hex.DecodeString("82AD2169626173652D6E616D6522F956402369626173652D756E6974200A24F9490025F9564000646E616D650164756E697406F95CB0036331323307F958B002F9514005F94900AA2169626173652D6E616D6522F956402369626173652D756E6974200A24F9490025F9564000646E616D6506F95CB007F958B005F94900")
require.Nil(t, err, "Decoding CBOR expected to succeed")
tr := senml.New(senml.CBOR)
msg := broker.Message{
Channel: "channel",
Subtopic: "subtopic",
Publisher: "publisher",
Protocol: "protocol",
Payload: cborBytes,
}
// 82AD2169626173652D6E616D6522F956402369626173652D756E6974200A24F9490025F9564000646E616D650164756E697406F95CB0036331323307F958B002F9514005F94900AA2169626173652D6E616D6522F956402369626173652D756E6974200A24F9490025F9564000646E616D6506F95CB007F958B005F94900
cborPld := msg
cborPld.ContentType = senml.CBOR
cborPld.Payload = cborBytes
tooManyMsg := msg
tooManyMsg.ContentType = senml.CBOR
tooManyMsg.Payload = tooManyBytes
val := 52.0
@ -80,18 +135,6 @@ func TestTransform(t *testing.T) {
msgs: msgs,
err: nil,
},
{
desc: "test normalize JSON",
msg: jsonPld,
msgs: msgs,
err: nil,
},
{
desc: "test normalize defaults to JSON",
msg: msg,
msgs: msgs,
err: nil,
},
{
desc: "test invalid payload",
msg: tooManyMsg,

25
writers/cassandra/README.md
View File

@ -8,17 +8,19 @@ The service is configured using the environment variables presented in the
following table. Note that any unset variables will be replaced with their
default values.
| Variable | Description | Default |
|--------------------------------------|-------------------------------------------------------------|------------------------|
| MF_NATS_URL | NATS instance URL | nats://localhost:4222 |
| MF_CASSANDRA_WRITER_LOG_LEVEL | Log level for Cassandra writer (debug, info, warn, error) | error |
| MF_CASSANDRA_WRITER_PORT | Service HTTP port | 8180 |
| MF_CASSANDRA_WRITER_DB_CLUSTER | Cassandra cluster comma separated addresses | 127.0.0.1 |
| MF_CASSANDRA_WRITER_DB_KEYSPACE | Cassandra keyspace name | mainflux |
| MF_CASSANDRA_WRITER_DB_USER | Cassandra DB username | |
| MF_CASSANDRA_WRITER_DB_PASS | Cassandra DB password | |
| MF_CASSANDRA_WRITER_DB_PORT | Cassandra DB port | 9042 |
| MF_CASSANDRA_WRITER_SUBJECTS_CONFIG | Configuration file path with subjects list | /config/subjects.toml |
| Variable | Description | Default |
|-------------------------------------|-----------------------------------------------------------|------------------------|
| MF_NATS_URL | NATS instance URL | nats://localhost:4222 |
| MF_CASSANDRA_WRITER_LOG_LEVEL | Log level for Cassandra writer (debug, info, warn, error) | error |
| MF_CASSANDRA_WRITER_PORT | Service HTTP port | 8180 |
| MF_CASSANDRA_WRITER_DB_CLUSTER | Cassandra cluster comma separated addresses | 127.0.0.1 |
| MF_CASSANDRA_WRITER_DB_KEYSPACE | Cassandra keyspace name | mainflux |
| MF_CASSANDRA_WRITER_DB_USER | Cassandra DB username | |
| MF_CASSANDRA_WRITER_DB_PASS | Cassandra DB password | |
| MF_CASSANDRA_WRITER_DB_PORT | Cassandra DB port | 9042 |
| MF_CASSANDRA_WRITER_SUBJECTS_CONFIG | Configuration file path with subjects list | /config/subjects.toml |
| MF_CASSANDRA_WRITER_CONTENT_TYPE | Message payload Content Type | application/senml+json |
## Deployment
```yaml
@ -39,6 +41,7 @@ default values.
MF_CASSANDRA_WRITER_DB_PASS: [Cassandra DB password]
MF_CASSANDRA_WRITER_DB_PORT: [Cassandra DB port]
MF_CASSANDRA_WRITER_SUBJECTS_CONFIG: [Configuration file path with subjects list]
MF_CASSANDRA_WRITER_CONTENT_TYPE: [Message payload Content Type]
ports:
- [host machine port]:[configured HTTP port]
volume:

2
writers/influxdb/README.md
View File

@ -19,6 +19,7 @@ default values.
| MF_INFLUX_WRITER_DB_PASS | Default password of InfluxDB user | mainflux |
| MF_INFLUX_WRITER_DB | InfluxDB database name | messages |
| MF_INFLUX_WRITER_SUBJECTS_CONFIG | Configuration file path with subjects list | /config/subjects.toml |
| MF_INFLUX_WRITER_CONTENT_TYPE | Message payload Content Type | application/senml+json |
## Deployment
@ -40,6 +41,7 @@ default values.
MF_INFLUX_WRITER_DB_USER: [InfluxDB admin user]
MF_INFLUX_WRITER_DB_PASS: [InfluxDB admin password]
MF_INFLUX_WRITER_SUBJECTS_CONFIG: [Configuration file path with subjects list]
MF_INFLUX_WRITER_CONTENT_TYPE: [Message payload Content Type]
ports:
- [host machine port]:[configured HTTP port]
volume:

2
writers/mongodb/README.md
View File

@ -17,6 +17,7 @@ default values.
| MF_MONGO_WRITER_DB_HOST | Default MongoDB database host | localhost |
| MF_MONGO_WRITER_DB_PORT | Default MongoDB database port | 27017 |
| MF_MONGO_WRITER_SUBJECTS_CONFIG | Configuration file path with subjects list | /config/subjects.toml |
| MF_MONGO_WRITER_CONTENT_TYPE | Message payload Content Type | application/senml+json |
## Deployment
@ -39,6 +40,7 @@ default values.
MF_MONGO_WRITER_DB_HOST: [MongoDB host]
MF_MONGO_WRITER_DB_PORT: [MongoDB port]
MF_MONGO_WRITER_SUBJETCS_CONFIG: [Configuration file path with subjects list]
MF_MONGO_WRITER_CONTENT_TYPE: [Message payload Content Type]
ports:
- [host machine port]:[configured HTTP port]
volume:

32
writers/postgres/README.md
View File

@ -8,21 +8,22 @@ The service is configured using the environment variables presented in the
following table. Note that any unset variables will be replaced with their
default values.
| Variable | Description | Default |
|-------------------------------------|---------------------------------------------|------------------------|
| MF_NATS_URL | NATS instance URL | nats://localhost:4222 |
| MF_POSTGRES_WRITER_LOG_LEVEL | Service log level | error |
| MF_POSTGRES_WRITER_PORT | Service HTTP port | 9104 |
| MF_POSTGRES_WRITER_DB_HOST | Postgres DB host | postgres |
| MF_POSTGRES_WRITER_DB_PORT | Postgres DB port | 5432 |
| MF_POSTGRES_WRITER_DB_USER | Postgres user | mainflux |
| MF_POSTGRES_WRITER_DB_PASS | Postgres password | mainflux |
| MF_POSTGRES_WRITER_DB | Postgres database name | messages |
| MF_POSTGRES_WRITER_DB_SSL_MODE | Postgres SSL mode | disabled |
| MF_POSTGRES_WRITER_DB_SSL_CERT | Postgres SSL certificate path | "" |
| MF_POSTGRES_WRITER_DB_SSL_KEY | Postgres SSL key | "" |
| MF_POSTGRES_WRITER_DB_SSL_ROOT_CERT | Postgres SSL root certificate path | "" |
| MF_POSTGRES_WRITER_SUBJECTS_CONFIG | Configuration file path with subjects list | /config/subjects.toml |
| Variable | Description | Default |
|-------------------------------------|--------------------------------------------|------------------------|
| MF_NATS_URL | NATS instance URL | nats://localhost:4222 |
| MF_POSTGRES_WRITER_LOG_LEVEL | Service log level | error |
| MF_POSTGRES_WRITER_PORT | Service HTTP port | 9104 |
| MF_POSTGRES_WRITER_DB_HOST | Postgres DB host | postgres |
| MF_POSTGRES_WRITER_DB_PORT | Postgres DB port | 5432 |
| MF_POSTGRES_WRITER_DB_USER | Postgres user | mainflux |
| MF_POSTGRES_WRITER_DB_PASS | Postgres password | mainflux |
| MF_POSTGRES_WRITER_DB | Postgres database name | messages |
| MF_POSTGRES_WRITER_DB_SSL_MODE | Postgres SSL mode | disabled |
| MF_POSTGRES_WRITER_DB_SSL_CERT | Postgres SSL certificate path | "" |
| MF_POSTGRES_WRITER_DB_SSL_KEY | Postgres SSL key | "" |
| MF_POSTGRES_WRITER_DB_SSL_ROOT_CERT | Postgres SSL root certificate path | "" |
| MF_POSTGRES_WRITER_SUBJECTS_CONFIG | Configuration file path with subjects list | /config/subjects.toml |
| MF_POSTGRES_WRITER_CONTENT_TYPE | Message payload Content Type | application/senml+json |
## Deployment
@ -49,6 +50,7 @@ default values.
MF_POSTGRES_WRITER_DB_SSL_KEY: [Postgres SSL key]
MF_POSTGRES_WRITER_DB_SSL_ROOT_CERT: [Postgres SSL Root cert]
MF_POSTGRES_WRITER_SUBJECTS_CONFIG: [Configuration file path with subjects list]
MF_POSTGRES_WRITER_CONTENT_TYPE: [Message payload Content Type]
ports:
- 9104:9104
networks: