backward compatible metadata keys

pkg/lib/envelope/metadata.go

@@ -7,8 +7,9 @@ import (
 
 // Metadata keys
 const (
-	KeyMessageType     = "Bacalhau-Type"
-	KeyPayloadEncoding = "Bacalhau-PayloadEncoding"
+	HeaderPrefix       = "Bacalhau-"
+	KeyMessageType     = "Type"
+	KeyPayloadEncoding = "PayloadEncoding"
 )
 
 // Metadata contains metadata about the message
@@ -23,11 +24,22 @@ func (m Metadata) ToMap() map[string]string {
 func (m Metadata) ToHeaders() map[string][]string {
 	headers := make(map[string][]string, len(m))
 	for k, v := range m {
-		headers[k] = []string{v}
+		headers[HeaderPrefix+k] = []string{v}
 	}
 	return headers
 }
 
+// FromHeaders creates a new Metadata object from a map[string][]string
+func FromHeaders(headers map[string][]string) *Metadata {
+	metadata := make(Metadata, len(headers))
+	for k, v := range headers {
+		if len(v) > 0 {
+			metadata[k[len(HeaderPrefix):]] = v[0]
+		}
+	}
+	return &metadata
+}
+
 // NewMetadataFromMap creates a new shallow copy Metadata object from a map.
 // Changes to the map will be reflected in the Metadata object, but more efficient than NewMetadataFromMapCopy
 func NewMetadataFromMap(m map[string]string) *Metadata {

pkg/lib/ncl/constants.go

@@ -2,9 +2,9 @@ package ncl
 
 // Metadata keys
 const (
-	KeySource      = "Bacalhau-Source"
-	KeyEventTime   = "Bacalhau-EventTime"
-	KeyMessageUUID = "Bacalhau-MessageUUID"
-	KeyMessageID   = "Bacalhau-MessageID"
-	KeySubject     = "Bacalhau-Subject"
+	KeySource      = "Source"
+	KeyEventTime   = "EventTime"
+	KeyMessageUUID = "MessageUUID"
+	KeyMessageID   = "MessageID"
+	KeySubject     = "Subject"
 )

pkg/lib/ncl/error_response.go

@@ -12,7 +12,7 @@ const (
 	ErrorMessageType = "ncl.ErrorResponse"
 
 	// KeyStatusCode is the key for the status code
-	KeyStatusCode = "Bacalhau-StatusCode"
+	KeyStatusCode = "StatusCode"
 
 	// StatusBadRequest is the status code for a bad request
 	StatusBadRequest = 400

pkg/lib/ncl/subscriber.go

@@ -75,7 +75,7 @@ func (s *subscriber) Subscribe(ctx context.Context, subjects ...string) error {
 // messageHandler is the callback function for message processing
 func (s *subscriber) handleNatsMessage(m *nats.Msg) {
 	if err := s.processMessage(m); err != nil {
-		log.Error().Err(err).Msg("failed to process message")
+		log.Error().Err(err).Str("handler", s.config.Name).Msg("failed to process message")
 
 		s.consecutiveFailures += 1
 		delay := s.config.Backoff.BackoffDuration(s.consecutiveFailures)

pkg/transport/nclprotocol/dispatcher/constants.go

@@ -1,5 +1,5 @@
 package dispatcher
 
 const (
-	KeySeqNum = "Bacalhau-SeqNum"
+	KeySeqNum = "SeqNum"
 )

pkg/transport/nclprotocol/types.go

@@ -53,7 +53,7 @@ type ConnectionHealth struct {
 }
 
 const (
-	KeySeqNum = "Bacalhau-SeqNum"
+	KeySeqNum = "SeqNum"
 )
 
 // MessageCreator defines how events from the watcher are converted into