gomessagestore/worker_getposition_test.go at master · blackhatbrigade/gomessagestore · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
package gomessagestore_test

import (
	"context"
	"io/ioutil"
	"testing"
	"time"

	. "github.com/blackhatbrigade/gomessagestore"
	"github.com/blackhatbrigade/gomessagestore/repository"
	mock_repository "github.com/blackhatbrigade/gomessagestore/repository/mocks"
	"github.com/blackhatbrigade/gomessagestore/uuid"
	"github.com/golang/mock/gomock"
	"github.com/sirupsen/logrus"
)

func TestSubscriberGetsPosition(t *testing.T) {

	tests := []struct {
		name             string
		subscriberID     string
		expectedError    error
		handlers         []MessageHandler
		expectedPosition int64
		expectedStream   string
		expectedCategory string
		opts             []SubscriberOption
		messages         []Message
		expectedHandled  []string
		positionEnvelope *repository.MessageEnvelope
	}{{
		name:             "When GetPosition is called (when no committed position exists) subscriber returns a position that matches the expected position",
		expectedPosition: 0,
		handlers:         []MessageHandler{&msgHandler{}},
		subscriberID:     "some id",
		opts: []SubscriberOption{
			SubscribeToEntityStream("some category", uuid1),
			SubscribeBatchSize(1),
		},
	}, {
		name:             "When GetPosition is called (for category) subscriber returns a position that matches the expected position",
		expectedPosition: 400,
		handlers:         []MessageHandler{&msgHandler{}},
		subscriberID:     "some id",
		opts: []SubscriberOption{
			SubscribeToCategory("some category"),
			SubscribeBatchSize(1),
		},
		positionEnvelope: &repository.MessageEnvelope{
			ID:             uuid.NewRandom(),
			StreamName:     "I_am_subscriber_id+position",
			StreamCategory: "I_am_subscriber_id+position",
			MessageType:    "PositionCommitted",
			Version:        5,
			GlobalPosition: 500,
			Data:           []byte("{\"position\":400}"),
			Time:           time.Unix(1, 5),
		},
	}, {
		name:             "When GetPosition is called (for event stream) subscriber returns a position that matches the expected position",
		expectedPosition: 400,
		handlers:         []MessageHandler{&msgHandler{}},
		subscriberID:     "some id",
		opts: []SubscriberOption{
			SubscribeToEntityStream("some category", uuid1),
			SubscribeBatchSize(1),
		},
		positionEnvelope: &repository.MessageEnvelope{
			ID:             uuid.NewRandom(),
			StreamName:     "I_am_subscriber_id+position",
			StreamCategory: "I_am_subscriber_id+position",
			MessageType:    "PositionCommitted",
			Version:        5,
			GlobalPosition: 500,
			Data:           []byte("{\"position\":400}"),
			Time:           time.Unix(1, 5),
		},
	}, {
		name:             "When GetPosition is called (for command stream) subscriber returns a position that matches the expected position",
		expectedPosition: 400,
		handlers:         []MessageHandler{&msgHandler{}},
		subscriberID:     "some id",
		opts: []SubscriberOption{
			SubscribeToCommandStream("some category"),
			SubscribeBatchSize(1),
		},
		positionEnvelope: &repository.MessageEnvelope{
			ID:             uuid.NewRandom(),
			StreamName:     "I_am_subscriber_id+position",
			StreamCategory: "I_am_subscriber_id+position",
			MessageType:    "PositionCommitted",
			Version:        5,
			GlobalPosition: 500,
			Data:           []byte("{\"position\":400}"),
			Time:           time.Unix(1, 5),
		},
	}}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			ctx := context.Background()
			mockRepo := mock_repository.NewMockRepository(ctrl)

			mockRepo.
				EXPECT().
				GetLastMessageInStream(ctx, "some id+position").
				Return(test.positionEnvelope, nil)

			var logrusLogger = logrus.New()
			logrusLogger.Out = ioutil.Discard
			myMessageStore := NewMessageStoreFromRepository(mockRepo, logrusLogger)

			defaultOptions := []SubscriberOption{SubscribeLogger(logrusLogger)}
			opts, err := GetSubscriberConfig(append(defaultOptions, test.opts...)...)
			panicIf(err)

			myWorker, err := CreateWorker(
				myMessageStore,
				"some id",
				test.handlers,
				opts,
			)

			if err != nil {
				t.Errorf("Failed on CreateWorker() Got: %s\n", err)
				return
			}

			pos, err := myWorker.GetPosition(ctx)

			if err != nil {
				t.Errorf("Failed on GetPosition() because of %v", err)
			}

			if pos != test.expectedPosition {
				t.Errorf("Failed on GetPosition()\n Expected%d\n Got: %d", test.expectedPosition, pos)
			}
		})
	}
}