Remove SetMessageEntry interface

gubled/gubled.go

@@ -71,16 +71,19 @@ var CreateMessageStore = func(args Args) store.MessageStore {
 	}
 }
 
-var CreateModules = func(router server.PubSubSource, args Args) []interface{} {
+var CreateModules = func(
+	router server.PubSubSource,
+	messageEntry server.MessageSink,
+	args Args) []interface{} {
 	modules := make([]interface{}, 0, 2)
 
-	if wsHandler, err := server.NewWSHandler(router, "/stream/"); err != nil {
+	if wsHandler, err := server.NewWSHandler(router, messageEntry, "/stream/"); err != nil {
 		guble.Err("Error loading WSHandler module: %s", err)
 	} else {
 		modules = append(modules, wsHandler)
 	}
 
-	modules = append(modules, server.NewRestMessageApi("/api/"))
+	modules = append(modules, server.NewRestMessageApi(messageEntry, "/api/"))
 
 	if args.GcmEnable {
 		if args.GcmApiKey == "" {
@@ -147,7 +150,7 @@ func StartupService(args Args) *server.Service {
 		accessManager,
 	)
 
-	for _, module := range CreateModules(router, args) {
+	for _, module := range CreateModules(router, messageEntry, args) {
 		service.Register(module)
 	}
 

gubled/gubled_test.go

@@ -140,14 +140,17 @@ func TestGcmOnlyStartedIfEnabled(t *testing.T) {
 	defer mockCtrl.Finish()
 
 	routerMock := NewMockPubSubSource(mockCtrl)
+
 	amMock := NewMockAccessManager(mockCtrl)
 	msMock := NewMockMessageStore(mockCtrl)
+	msinkMock := NewMockMessageSink(mockCtrl)
+
 	routerMock.EXPECT().KVStore().Return(store.NewMemoryKVStore(), nil)
 	routerMock.EXPECT().AccessManager().Return(amMock, nil).MaxTimes(2)
 	routerMock.EXPECT().MessageStore().Return(msMock, nil).MaxTimes(2)
 
-	a.True(containsGcmModule(CreateModules(routerMock, Args{GcmEnable: true, GcmApiKey: "xyz"})))
-	a.False(containsGcmModule(CreateModules(routerMock, Args{GcmEnable: false})))
+	a.True(containsGcmModule(CreateModules(routerMock, msinkMock, Args{GcmEnable: true, GcmApiKey: "xyz"})))
+	a.False(containsGcmModule(CreateModules(routerMock, msinkMock, Args{GcmEnable: false})))
 }
 
 func containsGcmModule(modules []interface{}) bool {
@@ -166,6 +169,8 @@ func TestPanicOnMissingGcmApiKey(t *testing.T) {
 	routerMock := NewMockPubSubSource(mockCtrl)
 	amMock := NewMockAccessManager(mockCtrl)
 	msMock := NewMockMessageStore(mockCtrl)
+	msinkMock := NewMockMessageSink(mockCtrl)
+
 	routerMock.EXPECT().AccessManager().Return(amMock, nil)
 	routerMock.EXPECT().MessageStore().Return(msMock, nil)
 
@@ -176,7 +181,7 @@ func TestPanicOnMissingGcmApiKey(t *testing.T) {
 		}
 	}()
 
-	CreateModules(routerMock, Args{GcmEnable: true})
+	CreateModules(routerMock, msinkMock, Args{GcmEnable: true})
 }
 
 func TestCreateStoreBackendPanicInvalidBackend(t *testing.T) {

gubled/mocks_server_gen_test.go

@@ -1,10 +1,11 @@
 // Automatically generated by MockGen. DO NOT EDIT!
-// Source: github.com/smancke/guble/server (interfaces: PubSubSource)
+// Source: github.com/smancke/guble/server (interfaces: PubSubSource,MessageSink)
 
 package gubled
 
 import (
 	gomock "github.com/golang/mock/gomock"
+	guble "github.com/smancke/guble/guble"
 	server "github.com/smancke/guble/server"
 	auth "github.com/smancke/guble/server/auth"
 	store "github.com/smancke/guble/store"
@@ -82,3 +83,34 @@ func (_m *MockPubSubSource) Unsubscribe(_param0 *server.Route) {
 func (_mr *_MockPubSubSourceRecorder) Unsubscribe(arg0 interface{}) *gomock.Call {
 	return _mr.mock.ctrl.RecordCall(_mr.mock, "Unsubscribe", arg0)
 }
+
+// Mock of MessageSink interface
+type MockMessageSink struct {
+	ctrl     *gomock.Controller
+	recorder *_MockMessageSinkRecorder
+}
+
+// Recorder for MockMessageSink (not exported)
+type _MockMessageSinkRecorder struct {
+	mock *MockMessageSink
+}
+
+func NewMockMessageSink(ctrl *gomock.Controller) *MockMessageSink {
+	mock := &MockMessageSink{ctrl: ctrl}
+	mock.recorder = &_MockMessageSinkRecorder{mock}
+	return mock
+}
+
+func (_m *MockMessageSink) EXPECT() *_MockMessageSinkRecorder {
+	return _m.recorder
+}
+
+func (_m *MockMessageSink) HandleMessage(_param0 *guble.Message) error {
+	ret := _m.ctrl.Call(_m, "HandleMessage", _param0)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+func (_mr *_MockMessageSinkRecorder) HandleMessage(arg0 interface{}) *gomock.Call {
+	return _mr.mock.ctrl.RecordCall(_mr.mock, "HandleMessage", arg0)
+}

scripts/generate_mocks.sh

@@ -25,7 +25,7 @@ MOCKGEN=$GOPATH/bin/mockgen
 $MOCKGEN  -self_package server -package server \
       -destination server/mocks_server_gen_test.go \
       github.com/smancke/guble/server \
-      PubSubSource,MessageSink,WSConnection,Startable,Stopable,SetMessageEntry,Endpoint
+      PubSubSource,MessageSink,WSConnection,Startable,Stopable,Endpoint
 replace "server/mocks_server_gen_test.go" "server \"github.com\/smancke\/guble\/server\"" "server\."
 
 $MOCKGEN -self_package server -package server \
@@ -61,7 +61,7 @@ $MOCKGEN -self_package gcm -package gcm \
 $MOCKGEN -package gubled \
       -destination gubled/mocks_server_gen_test.go \
       github.com/smancke/guble/server \
-      PubSubSource
+      PubSubSource,MessageSink
 
 $MOCKGEN -self_package gubled -package gubled \
       -destination gubled/mocks_auth_gen_test.go \

server/interfaces.go

@@ -76,11 +76,6 @@ type Stopable interface {
 	Stop() error
 }
 
-// SetMessageEntry interface for modules which need a MessageEntry set
-type SetMessageEntry interface {
-	SetMessageEntry(messageSink MessageSink)
-}
-
 // Endpoint adds a HTTP handler for the `GetPrefix()` to the webserver
 type Endpoint interface {
 	http.Handler

server/mocks_server_gen_test.go

@@ -1,5 +1,5 @@
 // Automatically generated by MockGen. DO NOT EDIT!
-// Source: github.com/smancke/guble/server (interfaces: PubSubSource,MessageSink,WSConnection,Startable,Stopable,SetMessageEntry,Endpoint)
+// Source: github.com/smancke/guble/server (interfaces: PubSubSource,MessageSink,WSConnection,Startable,Stopable,Endpoint)
 
 package server
 
@@ -227,35 +227,6 @@ func (_mr *_MockStopableRecorder) Stop() *gomock.Call {
 	return _mr.mock.ctrl.RecordCall(_mr.mock, "Stop")
 }
 
-// Mock of SetMessageEntry interface
-type MockSetMessageEntry struct {
-	ctrl     *gomock.Controller
-	recorder *_MockSetMessageEntryRecorder
-}
-
-// Recorder for MockSetMessageEntry (not exported)
-type _MockSetMessageEntryRecorder struct {
-	mock *MockSetMessageEntry
-}
-
-func NewMockSetMessageEntry(ctrl *gomock.Controller) *MockSetMessageEntry {
-	mock := &MockSetMessageEntry{ctrl: ctrl}
-	mock.recorder = &_MockSetMessageEntryRecorder{mock}
-	return mock
-}
-
-func (_m *MockSetMessageEntry) EXPECT() *_MockSetMessageEntryRecorder {
-	return _m.recorder
-}
-
-func (_m *MockSetMessageEntry) SetMessageEntry(_param0 MessageSink) {
-	_m.ctrl.Call(_m, "SetMessageEntry", _param0)
-}
-
-func (_mr *_MockSetMessageEntryRecorder) SetMessageEntry(arg0 interface{}) *gomock.Call {
-	return _mr.mock.ctrl.RecordCall(_mr.mock, "SetMessageEntry", arg0)
-}
-
 // Mock of Endpoint interface
 type MockEndpoint struct {
 	ctrl     *gomock.Controller

server/rest_message_api.go

@@ -15,14 +15,14 @@ import (
 const X_HEADER_PREFIX = "x-guble-"
 
 type RestMessageApi struct {
-	MessageSink MessageSink
-	mux         http.Handler
-	prefix      string
+	MessageSink
+	mux    http.Handler
+	prefix string
 }
 
-func NewRestMessageApi(prefix string) *RestMessageApi {
+func NewRestMessageApi(messageSink MessageSink, prefix string) *RestMessageApi {
 	mux := httprouter.New()
-	api := &RestMessageApi{mux: mux, prefix: prefix}
+	api := &RestMessageApi{messageSink, mux, prefix}
 
 	p := removeTrailingSlash(prefix)
 	mux.POST(p+"/message/*topic", api.PostMessage)
@@ -34,10 +34,6 @@ func (api *RestMessageApi) GetPrefix() string {
 	return api.prefix
 }
 
-func (api *RestMessageApi) SetMessageEntry(messageSink MessageSink) {
-	api.MessageSink = messageSink
-}
-
 func (api *RestMessageApi) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	api.mux.ServeHTTP(w, r)
 }

server/rest_message_api_test.go

@@ -22,8 +22,7 @@ func TestPostMessage(t *testing.T) {
 
 	// given:  a rest api with a message sink
 	messageSink := NewMockMessageSink(ctrl)
-	api := NewRestMessageApi("/api")
-	api.SetMessageEntry(messageSink)
+	api := NewRestMessageApi(messageSink, "/api")
 
 	url, _ := url.Parse("http://localhost/api/message/my/topic?userId=marvin&messageId=42")
 	// and a http context

server/service.go

@@ -53,7 +53,6 @@ func NewService(addr string, kvStore store.KVStore, messageStore store.MessageSt
 // does the expected tegistrations:
 //   Stopable: notify when the service stops
 //   Endpoint: Register the handler function of the Endpoint in the http service at prefix
-//   SetMessageEntry: Provide the message entry
 //
 // If the module does not have a HandlerFunc, the prefix parameter is ignored
 func (service *Service) Register(module interface{}) {
@@ -73,13 +72,6 @@ func (service *Service) Register(module interface{}) {
 		guble.Info("register %v as StopListener", name)
 		service.AddStopListener(m)
 	}
-
-	// do the injections ...
-
-	if m, ok := module.(SetMessageEntry); ok {
-		guble.Debug("inject MessageEntry to %v", name)
-		m.SetMessageEntry(service.messageSink)
-	}
 }
 
 func (service *Service) AddHandler(prefix string, handler http.Handler) {

server/service_test.go

@@ -49,20 +49,6 @@ func TestStopingOfModulesTimeout(t *testing.T) {
 	guble.Err(err.Error())
 }
 
-func TestRegistrationOfSetter(t *testing.T) {
-	defer initCtrl(t)()
-
-	// given:
-	service, _, _, messageSink, _ := aMockedService()
-	setMessageEntryMock := NewMockSetMessageEntry(ctrl)
-
-	// then I expect
-	setMessageEntryMock.EXPECT().SetMessageEntry(messageSink)
-
-	// when I register the modules
-	service.Register(setMessageEntryMock)
-}
-
 func TestEndpointRegisterAndServing(t *testing.T) {
 	defer initCtrl(t)()
 

server/websocket_connector.go

@@ -19,14 +19,17 @@ var webSocketUpgrader = websocket.Upgrader{
 }
 
 type WSHandler struct {
-	Router        PubSubSource
-	MessageSink   MessageSink
+	Router PubSubSource
+	MessageSink
 	prefix        string
 	messageStore  store.MessageStore
 	accessManager auth.AccessManager
 }
 
-func NewWSHandler(router PubSubSource, prefix string) (*WSHandler, error) {
+func NewWSHandler(
+	router PubSubSource,
+	messageSink MessageSink,
+	prefix string) (*WSHandler, error) {
 	accessManager, err := router.AccessManager()
 	if err != nil {
 		return nil, err
@@ -39,6 +42,7 @@ func NewWSHandler(router PubSubSource, prefix string) (*WSHandler, error) {
 
 	return &WSHandler{
 		Router:        router,
+		MessageSink:   messageSink,
 		prefix:        prefix,
 		accessManager: accessManager,
 		messageStore:  messageStore,
@@ -49,10 +53,6 @@ func (handle *WSHandler) GetPrefix() string {
 	return handle.prefix
 }
 
-func (handle *WSHandler) SetMessageEntry(messageSink MessageSink) {
-	handle.MessageSink = messageSink
-}
-
 func (handle *WSHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	c, err := webSocketUpgrader.Upgrade(w, r, nil)
 	if err != nil {