Added additional error condition tests

eventstore_test.go

@@ -361,6 +361,144 @@ func TestReadAllCancelsSubscriptionWithClose(t *testing.T) {
 	}
 }
 
+type enc struct {
+	encode func(interface{}) (*eventstore.EncodedEvt, error)
+	decode func(*eventstore.EncodedEvt) (interface{}, error)
+}
+
+func (e enc) Encode(evt interface{}) (*eventstore.EncodedEvt, error) {
+	return e.encode(evt)
+}
+
+func (e enc) Decode(evt *eventstore.EncodedEvt) (interface{}, error) {
+	return e.decode(evt)
+}
+
+func TestEncoderEncodeErrorsPropagated(t *testing.T) {
+	if !*integration {
+		t.Skip("skipping integration tests")
+	}
+
+	var anErr = fmt.Errorf("an error occurred")
+
+	e := enc{
+		encode: func(i interface{}) (*eventstore.EncodedEvt, error) { return nil, anErr },
+	}
+
+	es, cleanup := eventStoreWithDec(t, e)
+
+	defer cleanup()
+
+	err := es.AppendStream(
+		context.Background(),
+		"stream",
+		eventstore.InitialStreamVersion,
+		[]interface{}{
+			SomeEvent{
+				UserID: "123",
+			},
+		},
+	)
+
+	if !errors.Is(err, anErr) {
+		t.Fatal("error should have been propagated")
+	}
+}
+
+func TestEncoderDecodeErrorsPropagated(t *testing.T) {
+	if !*integration {
+		t.Skip("skipping integration tests")
+	}
+
+	var anErr = fmt.Errorf("an error occurred")
+
+	e := enc{
+		encode: func(i interface{}) (*eventstore.EncodedEvt, error) {
+			return &eventstore.EncodedEvt{
+				Data: "malformed-json",
+				Type: "foo",
+			}, nil
+		},
+		decode: func(ee *eventstore.EncodedEvt) (interface{}, error) {
+			return nil, anErr
+		},
+	}
+
+	es, cleanup := eventStoreWithDec(t, e)
+
+	defer cleanup()
+
+	err := es.AppendStream(
+		context.Background(),
+		"stream",
+		eventstore.InitialStreamVersion,
+		[]interface{}{
+			SomeEvent{
+				UserID: "123",
+			},
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = es.ReadStream(context.Background(), "stream")
+
+	if !errors.Is(err, anErr) {
+		t.Fatal("error should have been propagated")
+	}
+}
+
+func TestEncoderDecodeErrorsPropagatedOnReadAll(t *testing.T) {
+	if !*integration {
+		t.Skip("skipping integration tests")
+	}
+
+	var anErr = fmt.Errorf("an error occurred")
+
+	e := enc{
+		encode: func(i interface{}) (*eventstore.EncodedEvt, error) {
+			return &eventstore.EncodedEvt{
+				Data: "malformed-json",
+				Type: "foo",
+			}, nil
+		},
+		decode: func(ee *eventstore.EncodedEvt) (interface{}, error) {
+			return nil, anErr
+		},
+	}
+
+	es, cleanup := eventStoreWithDec(t, e)
+
+	defer cleanup()
+
+	err := es.AppendStream(
+		context.Background(),
+		"stream",
+		eventstore.InitialStreamVersion,
+		[]interface{}{
+			SomeEvent{
+				UserID: "123",
+			},
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sub, err := es.ReadAll(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	defer sub.Close()
+
+	err = <-sub.Err
+
+	if !errors.Is(err, anErr) {
+		t.Fatal("error should have been propagated")
+	}
+}
 func TestNewEncoderMustBeProvided(t *testing.T) {
 	_, err := eventstore.New("foo", nil)
 	if err == nil {
@@ -436,12 +574,16 @@ func TestReadStreamValidation(t *testing.T) {
 }
 
 func eventStore(t *testing.T) (*eventstore.EventStore, func()) {
+	return eventStoreWithDec(t, eventstore.NewJSONEncoder(SomeEvent{}))
+}
+
+func eventStoreWithDec(t *testing.T, enc eventstore.Encoder) (*eventstore.EventStore, func()) {
 	file, err := os.CreateTemp(os.TempDir(), "es-db-*")
 	if err != nil {
 		t.Fatalf("could not create tem file: %v", err)
 	}
 
-	es, err := eventstore.New(file.Name(), eventstore.NewJsonEncoder(SomeEvent{}))
+	es, err := eventstore.New(file.Name(), enc)
 	if err != nil {
 		t.Fatalf("error creating es: %v", err)
 	}

example/cmd/api/main.go

@@ -12,7 +12,7 @@ import (
 func main() {
 	estore, err := eventstore.New(
 		"exampledb",
-		eventstore.NewJsonEncoder(
+		eventstore.NewJSONEncoder(
 			account.NewAccountOpenned{},
 		),
 	)

example/cmd/projections/main.go

@@ -14,7 +14,7 @@ import (
 func main() {
 	estore, err := eventstore.New(
 		"exampledb",
-		eventstore.NewJsonEncoder(
+		eventstore.NewJSONEncoder(
 			account.NewAccountOpenned{},
 		),
 	)

json_encoder.go

@@ -2,11 +2,13 @@ package eventstore
 
 import (
 	"encoding/json"
+	"fmt"
 	"reflect"
 )
 
-// NewJsonEncoder constructs json encoder
-func NewJsonEncoder(evts ...interface{}) *JsonEncoder {
+// NewJSONEncoder constructs json encoder
+// It receives a slice of event types it should be able to encode/decode
+func NewJSONEncoder(evts ...interface{}) *JsonEncoder {
 	enc := JsonEncoder{
 		types: make(map[string]reflect.Type),
 	}
@@ -26,21 +28,26 @@ type JsonEncoder struct {
 }
 
 // Encode marshals incoming event to it's json representation
-func (e *JsonEncoder) Encode(evtData interface{}) (*EncodedEvt, error) {
-	data, err := json.Marshal(evtData)
+func (e *JsonEncoder) Encode(evt interface{}) (*EncodedEvt, error) {
+	data, err := json.Marshal(evt)
 	if err != nil {
 		return nil, err
 	}
 
 	return &EncodedEvt{
-		Type: reflect.TypeOf(evtData).Name(),
+		Type: reflect.TypeOf(evt).Name(),
 		Data: string(data),
 	}, nil
 }
 
 // Decode unmarshals incoming event to it's corresponding go type
 func (e *JsonEncoder) Decode(evt *EncodedEvt) (interface{}, error) {
-	v := reflect.New(e.types[evt.Type])
+	t, ok := e.types[evt.Type]
+	if !ok {
+		return nil, fmt.Errorf("event not registered via NewJSONEncoder")
+	}
+
+	v := reflect.New(t)
 
 	err := json.Unmarshal([]byte(evt.Data), v.Interface())
 	if err != nil {

json_encoder_test.go

@@ -12,7 +12,7 @@ type AnotherEvent struct {
 }
 
 func TestShouldDecodeEncodedEvent(t *testing.T) {
-	enc := eventstore.NewJsonEncoder(SomeEvent{}, AnotherEvent{})
+	enc := eventstore.NewJSONEncoder(SomeEvent{}, AnotherEvent{})
 
 	decodeEncode(t, enc, SomeEvent{
 		UserID: "some-user",
@@ -38,3 +38,27 @@ func decodeEncode(t *testing.T, enc eventstore.Encoder, e interface{}) {
 		t.Fatalf("event not decoded. want: %#v, got: %#v", e, decoded)
 	}
 }
+
+func TestShouldErrorOutIfMalformedJSON(t *testing.T) {
+	enc := eventstore.NewJSONEncoder(SomeEvent{}, AnotherEvent{})
+
+	_, err := enc.Decode(&eventstore.EncodedEvt{
+		Data: "malformed-json",
+		Type: "SomeEvent",
+	})
+	if err == nil {
+		t.Fatal("should error out")
+	}
+}
+
+func TestShouldErrorOutIfEvtTypeUnknown(t *testing.T) {
+	enc := eventstore.NewJSONEncoder(SomeEvent{}, AnotherEvent{})
+
+	_, err := enc.Decode(&eventstore.EncodedEvt{
+		Data: `{"userId": "123"}`,
+		Type: "unknown",
+	})
+	if err == nil {
+		t.Fatal("should error out")
+	}
+}