feat(message): Add Flag Support (#264)

CONTRIBUTING.md

@@ -13,6 +13,7 @@ Thank you so much for your interest in contributing to Substation! This document
 
 [Development](#development)
   + [Development Environment](#development-environment)
+  + [Messages](#messages)
   + [Conditions](#conditions)
   + [Transforms](#transforms)
   + [Testing](#testing)
@@ -48,6 +49,12 @@ Enhancements should be submitted as issues using the issue template.
 
 The project supports development through the use of [Visual Studio Code configurations](https://code.visualstudio.com/docs/remote/containers). The VS Code [development container](.devcontainer/Dockerfile) contains all packages required to develop and test changes locally before submitting pull requests.
 
+### [Messages](message/)
+
+Each message can have a series of flags attached to it that are used to determine how the message should be processed by the system. These flags are exported as iota constants and should use verb style naming, such as:
+- `IsControl`
+- `SkipMissingValues`
+
 ### [Conditions](condition/)
 
 Each condition should be functional and solve a single problem, and each one is nested under a "family" of conditions. (We may ask that you split complex condition logic into multiple conditions.) For example, there is a family for string comparisons:

cmd/aws/lambda/substation/api_gateway.go

@@ -49,7 +49,7 @@ func gatewayHandler(ctx context.Context, request events.APIGatewayProxyRequest)
 
 	b := []byte(request.Body)
 	msg := []*message.Message{
-		message.New().SetData(b).SetMetadata(metadata),
+		message.New().SetData(b).SetMetadata(metadata).SkipMissingValues(),
 		message.New().AsControl(),
 	}
 
@@ -61,7 +61,7 @@ func gatewayHandler(ctx context.Context, request events.APIGatewayProxyRequest)
 	// Convert transformed messages to a JSON array.
 	var output []json.RawMessage
 	for _, msg := range res {
-		if msg.IsControl() {
+		if msg.HasFlag(message.IsControl) {
 			continue
 		}
 

cmd/aws/lambda/substation/data_firehose.go

@@ -48,7 +48,7 @@ func firehoseHandler(ctx context.Context, event events.KinesisFirehoseEvent) (ev
 			return resp, err
 		}
 
-		msg := message.New().SetData(record.Data).SetMetadata(metadata)
+		msg := message.New().SetData(record.Data).SetMetadata(metadata).SkipMissingValues()
 		res, err := sub.Transform(ctx, msg)
 		if err != nil {
 			return resp, err

cmd/aws/lambda/substation/dynamodb.go

@@ -142,7 +142,7 @@ func dynamodbHandler(ctx context.Context, event events.DynamoDBEvent) error {
 			//   "before": { ... },
 			//   "after": { ... }
 			// }
-			msg := message.New().SetMetadata(metadata)
+			msg := message.New().SetMetadata(metadata).SkipMissingValues()
 			if err := msg.SetValue("source.ts_ms", record.Change.ApproximateCreationDateTime.Time.UnixMilli()); err != nil {
 				return err
 			}

cmd/aws/lambda/substation/kinesis_data_stream.go

@@ -113,7 +113,7 @@ func kinesisStreamHandler(ctx context.Context, event events.KinesisEvent) error
 				return err
 			}
 
-			msg := message.New().SetData(record.Data).SetMetadata(metadata)
+			msg := message.New().SetData(record.Data).SetMetadata(metadata).SkipMissingValues()
 			ch.Send(msg)
 		}
 

cmd/aws/lambda/substation/lambda.go

@@ -32,7 +32,7 @@ func lambdaHandler(ctx context.Context, event json.RawMessage) ([]json.RawMessag
 
 	// Data and ctrl messages are sent as a group.
 	msg := []*message.Message{
-		message.New().SetData(evt),
+		message.New().SetData(evt).SkipMissingValues(),
 		message.New().AsControl(),
 	}
 
@@ -44,7 +44,7 @@ func lambdaHandler(ctx context.Context, event json.RawMessage) ([]json.RawMessag
 	// Convert transformed messages to a JSON array.
 	var output []json.RawMessage
 	for _, msg := range res {
-		if msg.IsControl() {
+		if msg.HasFlag(message.IsControl) {
 			continue
 		}
 

cmd/aws/lambda/substation/s3.go

@@ -159,7 +159,7 @@ func s3Handler(ctx context.Context, event events.S3Event) error {
 					return err
 				}
 
-				msg := message.New().SetData(r).SetMetadata(metadata)
+				msg := message.New().SetData(r).SetMetadata(metadata).SkipMissingValues()
 				ch.Send(msg)
 
 				return nil
@@ -180,7 +180,7 @@ func s3Handler(ctx context.Context, event events.S3Event) error {
 				}
 
 				b := []byte(scanner.Text())
-				msg := message.New().SetData(b).SetMetadata(metadata)
+				msg := message.New().SetData(b).SetMetadata(metadata).SkipMissingValues()
 
 				ch.Send(msg)
 			}
@@ -336,7 +336,7 @@ func s3SnsHandler(ctx context.Context, event events.SNSEvent) error {
 						return err
 					}
 
-					msg := message.New().SetData(r).SetMetadata(metadata)
+					msg := message.New().SetData(r).SetMetadata(metadata).SkipMissingValues()
 					ch.Send(msg)
 
 					return nil
@@ -357,7 +357,7 @@ func s3SnsHandler(ctx context.Context, event events.SNSEvent) error {
 					}
 
 					b := []byte(scanner.Text())
-					msg := message.New().SetData(b).SetMetadata(metadata)
+					msg := message.New().SetData(b).SetMetadata(metadata).SkipMissingValues()
 
 					ch.Send(msg)
 				}

cmd/aws/lambda/substation/sns.go

@@ -100,7 +100,7 @@ func snsHandler(ctx context.Context, event events.SNSEvent) error {
 
 		for _, record := range event.Records {
 			b := []byte(record.SNS.Message)
-			msg := message.New().SetData(b).SetMetadata(metadata)
+			msg := message.New().SetData(b).SetMetadata(metadata).SkipMissingValues()
 
 			ch.Send(msg)
 		}

cmd/aws/lambda/substation/sqs.go

@@ -99,7 +99,7 @@ func sqsHandler(ctx context.Context, event events.SQSEvent) error {
 
 		for _, record := range event.Records {
 			b := []byte(record.Body)
-			msg := message.New().SetData(b).SetMetadata(metadata)
+			msg := message.New().SetData(b).SetMetadata(metadata).SkipMissingValues()
 
 			ch.Send(msg)
 		}

cmd/development/substation-bench/main.go

@@ -170,7 +170,7 @@ func main() {
 		defer ch.Close()
 
 		for _, b := range data {
-			msg := message.New().SetData(b)
+			msg := message.New().SetData(b).SkipMissingValues()
 			ch.Send(msg)
 		}
 

cmd/development/substation-file/main.go

@@ -151,7 +151,7 @@ func run(ctx context.Context, opts options) error {
 				return err
 			}
 
-			msg := message.New().SetData(r)
+			msg := message.New().SetData(r).SkipMissingValues()
 			ch.Send(msg)
 
 			return nil
@@ -172,7 +172,7 @@ func run(ctx context.Context, opts options) error {
 			}
 
 			b := []byte(scanner.Text())
-			msg := message.New().SetData(b)
+			msg := message.New().SetData(b).SkipMissingValues()
 
 			ch.Send(msg)
 		}

cmd/development/substation-kinesis-tap/main.go

@@ -237,7 +237,7 @@ func run(ctx context.Context, opts options) error {
 					log.WithField("stream", opts.StreamName).WithField("shard", shard.ShardId).WithField("count", len(deagg)).Debug("Retrieved records from Kinesis shard.")
 
 					for _, record := range deagg {
-						msg := message.New().SetData(record.Data)
+						msg := message.New().SetData(record.Data).SkipMissingValues()
 						ch.Send(msg)
 					}
 

cmd/substation/demo.go

@@ -326,7 +326,7 @@ partially normalized to the Elastic Common Schema (ECS).
 		}
 
 		msgs := []*message.Message{
-			message.New().SetData([]byte(demoEvt)),
+			message.New().SetData([]byte(demoEvt)).SkipMissingValues(),
 			message.New().AsControl(),
 		}
 

cmd/substation/test.go

@@ -281,7 +281,7 @@ func testFile(arg string, extVars map[string]string) error {
 
 		for _, msg := range tMsgs {
 			// Skip control messages because they contain no data.
-			if msg.IsControl() {
+			if msg.HasFlag(message.IsControl) {
 				continue
 			}
 

condition/format_json.go

@@ -34,7 +34,7 @@ type formatJSON struct {
 }
 
 func (c *formatJSON) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/format_mime.go

@@ -53,7 +53,7 @@ type formatMIME struct {
 }
 
 func (c *formatMIME) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/network_ip_global_unicast.go

@@ -27,7 +27,7 @@ type networkIPGlobalUnicast struct {
 }
 
 func (insp *networkIPGlobalUnicast) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/network_ip_link_local_multicast.go

@@ -27,7 +27,7 @@ type networkIPLinkLocalMulticast struct {
 }
 
 func (insp *networkIPLinkLocalMulticast) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/network_ip_link_local_unicast.go

@@ -27,7 +27,7 @@ type networkIPLinkLocalUnicast struct {
 }
 
 func (insp *networkIPLinkLocalUnicast) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/network_ip_loopback.go

@@ -27,7 +27,7 @@ type networkIPLoopback struct {
 }
 
 func (insp *networkIPLoopback) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/network_ip_multicast.go

@@ -27,7 +27,7 @@ type networkIPMulticast struct {
 }
 
 func (insp *networkIPMulticast) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/network_ip_private.go

@@ -27,7 +27,7 @@ type networkIPPrivate struct {
 }
 
 func (insp *networkIPPrivate) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/network_ip_unicast.go

@@ -27,7 +27,7 @@ type networkIPUnicast struct {
 }
 
 func (insp *networkIPUnicast) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/network_ip_unspecified.go

@@ -27,7 +27,7 @@ type networkIPUnspecified struct {
 }
 
 func (insp *networkIPUnspecified) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/network_ip_valid.go

@@ -27,7 +27,7 @@ type networkIPValid struct {
 }
 
 func (insp *networkIPValid) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/number_bitwise_and.go

@@ -26,7 +26,7 @@ type numberBitwiseAND struct {
 }
 
 func (insp *numberBitwiseAND) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/number_bitwise_not.go

@@ -26,7 +26,7 @@ type numberBitwiseNOT struct {
 }
 
 func (insp *numberBitwiseNOT) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/number_bitwise_or.go

@@ -26,7 +26,7 @@ type numberBitwiseOR struct {
 }
 
 func (insp *numberBitwiseOR) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/number_bitwise_xor.go

@@ -26,7 +26,7 @@ type numberBitwiseXOR struct {
 }
 
 func (insp *numberBitwiseXOR) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/number_equal_to.go

@@ -25,7 +25,7 @@ type numberEqualTo struct {
 }
 
 func (insp *numberEqualTo) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 	compare := insp.conf.Value
@@ -40,7 +40,6 @@ func (insp *numberEqualTo) Condition(ctx context.Context, msg *message.Message)
 	}
 
 	target := msg.GetValue(insp.conf.Object.TargetKey)
-
 	if target.Exists() {
 		compare = target.Float()
 	}

condition/number_greater_than.go

@@ -27,7 +27,7 @@ type numberGreaterThan struct {
 }
 
 func (insp *numberGreaterThan) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/number_length_equal_to.go

@@ -26,7 +26,7 @@ type numberLengthEqualTo struct {
 }
 
 func (insp *numberLengthEqualTo) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/number_length_greater_than.go

@@ -26,7 +26,7 @@ type numberLengthGreaterThan struct {
 }
 
 func (insp *numberLengthGreaterThan) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/number_length_less_than.go

@@ -26,7 +26,7 @@ type numberLengthLessThan struct {
 }
 
 func (insp *numberLengthLessThan) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/number_less_than.go

@@ -25,7 +25,7 @@ type numberLessThan struct {
 }
 
 func (insp *numberLessThan) Condition(ctx context.Context, msg *message.Message) (output bool, err error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}
 

condition/string_contains.go

@@ -30,7 +30,7 @@ type stringContains struct {
 }
 
 func (insp *stringContains) Condition(ctx context.Context, msg *message.Message) (bool, error) {
-	if msg.IsControl() {
+	if msg.HasFlag(message.IsControl) {
 		return false, nil
 	}