fix: skip message when invalid JSON

message/handler/validate.go

@@ -19,7 +19,8 @@ func ValidateMessage(validator message.Validator, failWhenInvalid bool) watermil
  return func(msg *watermillmessage.Message) ([]*watermillmessage.Message, error) {
  validationErr, err := validator(msg)
  if err != nil {
- return nil, err
+ logrus.WithError(err).Error("failed to validate message. Skipping message...")
+ return nil, nil
  }
 
  if validationErr != nil {

message/handler/validate_test.go

@@ -6,6 +6,7 @@ import (
 
  watermillmessage "github.com/ThreeDotsLabs/watermill/message"
  "github.com/asyncapi/event-gateway/message"
+ "github.com/pkg/errors"
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
 )
@@ -17,6 +18,7 @@ func TestValidateMessage(t *testing.T) {
  failWhenInvalid bool
  expectedValidationErr string
  expectedErr error
+ messageIsInvalid bool
  }{
  {
  name: "Message is valid",
@@ -36,6 +38,11 @@ func TestValidateMessage(t *testing.T) {
  expectedValidationErr: "testing error",
  expectedErr: ErrMessageIsInvalid,
  },
+ {
+ name: "Skip message if message is invalid",
+ validator: erroredMessageValidator,
+ messageIsInvalid: true,
+ },
  }
  for _, test := range tests {
  t.Run(test.name, func(t *testing.T) {
@@ -47,8 +54,12 @@ func TestValidateMessage(t *testing.T) {
  assert.NoError(t, err)
  }
 
- assert.Len(t, returnedMsgs, 1)
- assert.Same(t, msg, returnedMsgs[0])
+ if test.messageIsInvalid {
+ assert.Empty(t, returnedMsgs)
+ } else {
+ assert.Len(t, returnedMsgs, 1)
+ assert.Same(t, msg, returnedMsgs[0])
+ }
 
  validationErr, err := message.ValidationErrorFromMessage(msg)
  require.NoError(t, err)
@@ -66,3 +77,7 @@ func invalidMessageValidator(_ *watermillmessage.Message) (*message.ValidationEr
  Errors: []string{"testing error"},
  }, nil
 }
+
+func erroredMessageValidator(_ *watermillmessage.Message) (*message.ValidationError, error) {
+ return nil, errors.New("random error")
+}