All tests operational again good to go.

Slowly cleaning out the cruft, making it useful for all the things.

Signed-off-by: quobix <[email protected]>

bridge/broker_connector_test.go

@@ -114,14 +114,6 @@ func TestBrokerConnector_BadConfig(t *testing.T) {
  "Missing address from config",
  &BrokerConnectorConfig{Username: "guest", Password: "guest"},
  fmt.Errorf("config invalid, config missing server address")},
- {
- "Missing username from config",
- &BrokerConnectorConfig{ServerAddr: "somewhere:000"},
- fmt.Errorf("config invalid, config missing username")},
- {
- "Missing password from config",
- &BrokerConnectorConfig{Username: "hi", ServerAddr: "somewhere:000"},
- fmt.Errorf("config invalid, config missing password")},
  }
 
  for _, tc := range tt {

bus/channel_test.go

@@ -4,6 +4,7 @@
 package bus
 
 import (
+ "context"
  "github.com/go-stomp/stomp/v3/frame"
  "github.com/google/uuid"
  "github.com/pb33f/ranch/bridge"
@@ -247,6 +248,16 @@ type MockBridgeConnection struct {
  Id *uuid.UUID
 }
 
+func (c *MockBridgeConnection) Conversation(destination string, payload []byte, opts ...func(*frame.Frame) error) (bridge.Subscription, error) {
+ //TODO implement me
+ panic("implement me")
+}
+
+func (c *MockBridgeConnection) RequestResponse(ctx context.Context, payload []byte, opts ...func(*frame.Frame) error) (*model.Message, error) {
+ //TODO implement me
+ panic("implement me")
+}
+
 func (c *MockBridgeConnection) GetId() *uuid.UUID {
  return c.Id
 }

bus/eventbus.go

@@ -462,8 +462,8 @@ func (bus *transportEventBus) StartFabricEndpoint(
  connectionListener stompserver.RawConnectionListener, config EndpointConfig) error {
 
  if bus.fabEndpoint != nil {
- return nil
- // return fmt.Errorf("unable to start: fabric endpoint is already running")
+ //return nil
+ return fmt.Errorf("unable to start: fabric endpoint is already running")
  }
  if configErr := config.validate(); configErr != nil {
  return configErr

bus/fabric_endpoint_test.go

@@ -347,9 +347,9 @@ func TestFabricEndpoint_BridgeMessage(t *testing.T) {
 
  id1 := uuid.New()
  req1, _ := json.Marshal(model.Request{
- Request: "test-request",
- Payload: "test-rq",
- Id: &id1,
+ RequestCommand: "test-request",
+ Payload:  "test-rq",
+ Id:  &id1,
  })
 
  wg.Add(1)
@@ -363,9 +363,9 @@ func TestFabricEndpoint_BridgeMessage(t *testing.T) {
 
  id2 := uuid.New()
  req2, _ := json.Marshal(model.Request{
- Request: "test-request2",
- Payload: "test-rq2",
- Id: &id2,
+ RequestCommand: "test-request2",
+ Payload:  "test-rq2",
+ Id:  &id2,
  })
 
  wg.Wait()
@@ -378,14 +378,14 @@ func TestFabricEndpoint_BridgeMessage(t *testing.T) {
 
  receivedReq := messages[0].Payload.(*model.Request)
 
- assert.Equal(t, receivedReq.Request, "test-request")
+ assert.Equal(t, receivedReq.RequestCommand, "test-request")
  assert.Equal(t, receivedReq.Payload, "test-rq")
  assert.Equal(t, *receivedReq.Id, id1)
  assert.Nil(t, receivedReq.BrokerDestination)
 
  receivedReq2 := messages[1].Payload.(*model.Request)
 
- assert.Equal(t, receivedReq2.Request, "test-request2")
+ assert.Equal(t, receivedReq2.RequestCommand, "test-request2")
  assert.Equal(t, receivedReq2.Payload, "test-rq2")
  assert.Equal(t, *receivedReq2.Id, id2)
  assert.Equal(t, receivedReq2.BrokerDestination.ConnectionId, "con2")

bus/store_sync_service_test.go

@@ -53,16 +53,16 @@ func TestStoreSyncService_OpenStoreErrors(t *testing.T) {
  id := uuid.New()
  bus.SendRequestMessage(syncChan, "invalid-request", nil)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: openStoreRequest,
- Payload: "invalid-payload",
- Id: &id,
+ RequestCommand: openStoreRequest,
+ Payload:  "invalid-payload",
+ Id:  &id,
  }, nil)
 
  wg.Add(1)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: openStoreRequest,
- Payload: make(map[string]interface{}),
- Id: &id,
+ RequestCommand: openStoreRequest,
+ Payload:  make(map[string]interface{}),
+ Id:  &id,
  }, nil)
  wg.Wait()
 
@@ -72,9 +72,9 @@ func TestStoreSyncService_OpenStoreErrors(t *testing.T) {
 
  wg.Add(1)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: openStoreRequest,
- Payload: map[string]interface{}{"storeId": "non-existing-store"},
- Id: &id,
+ RequestCommand: openStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "non-existing-store"},
+ Id:  &id,
  }, nil)
  wg.Wait()
 
@@ -111,8 +111,8 @@ func TestStoreSyncService_OpenStore(t *testing.T) {
 
  wg.Add(1)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: openStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
+ RequestCommand: openStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
  }, nil)
  wg.Wait()
 
@@ -135,8 +135,8 @@ func TestStoreSyncService_OpenStore(t *testing.T) {
 
  wg.Add(1)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: openStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
+ RequestCommand: openStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
  }, nil)
  wg.Wait()
 
@@ -157,8 +157,8 @@ func TestStoreSyncService_OpenStore(t *testing.T) {
 
  wg.Add(1)
  bus.SendRequestMessage(syncChan2, &model.Request{
- Request: openStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
+ RequestCommand: openStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
  }, nil)
  wg.Wait()
 
@@ -230,33 +230,33 @@ func TestStoreSyncService_CloseStore(t *testing.T) {
 
  wg.Add(2)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: openStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
+ RequestCommand: openStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
  }, nil)
  bus.SendRequestMessage(syncChan2, &model.Request{
- Request: openStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
+ RequestCommand: openStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
  }, nil)
  wg.Wait()
 
  assert.Equal(t, len(service.syncStoreListeners["test-store"].clientSyncChannels), 2)
 
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: closeStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
- Id: &id,
+ RequestCommand: closeStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
+ Id:  &id,
  }, nil)
 
  wg.Add(2)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: closeStoreRequest,
- Payload: make(map[string]interface{}),
- Id: &id,
+ RequestCommand: closeStoreRequest,
+ Payload:  make(map[string]interface{}),
+ Id:  &id,
  }, nil)
  bus.SendRequestMessage(syncChan2, &model.Request{
- Request: closeStoreRequest,
- Payload: map[string]interface{}{"storeId": ""},
- Id: &id,
+ RequestCommand: closeStoreRequest,
+ Payload:  map[string]interface{}{"storeId": ""},
+ Id:  &id,
  }, nil)
  wg.Wait()
 
@@ -272,21 +272,21 @@ func TestStoreSyncService_CloseStore(t *testing.T) {
  service.lock.Unlock()
 
  bus.SendRequestMessage(syncChan2, &model.Request{
- Request: closeStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
- Id: &id,
+ RequestCommand: closeStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
+ Id:  &id,
  }, nil)
 
  wg.Add(2)
  bus.SendRequestMessage(syncChan2, &model.Request{
- Request: closeStoreRequest,
- Payload: make(map[string]interface{}),
- Id: &id,
+ RequestCommand: closeStoreRequest,
+ Payload:  make(map[string]interface{}),
+ Id:  &id,
  }, nil)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: closeStoreRequest,
- Payload: map[string]interface{}{"storeId": ""},
- Id: &id,
+ RequestCommand: closeStoreRequest,
+ Payload:  map[string]interface{}{"storeId": ""},
+ Id:  &id,
  }, nil)
  wg.Wait()
 
@@ -323,29 +323,29 @@ func TestStoreSyncService_UpdateStoreErrors(t *testing.T) {
 
  wg.Add(1)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: updateStoreRequest,
- Payload: map[string]interface{}{},
- Id: &id,
+ RequestCommand: updateStoreRequest,
+ Payload:  map[string]interface{}{},
+ Id:  &id,
  }, nil)
  wg.Wait()
 
  assert.Equal(t, syncResp[0].(*model.Response).ErrorMessage, "Invalid UpdateStoreRequest: missing storeId")
 
  wg.Add(1)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: updateStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
- Id: &id,
+ RequestCommand: updateStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
+ Id:  &id,
  }, nil)
  wg.Wait()
 
  assert.Equal(t, syncResp[1].(*model.Response).ErrorMessage, "Invalid UpdateStoreRequest: missing itemId")
 
  wg.Add(1)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: updateStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store", "itemId": "item1"},
- Id: &id,
+ RequestCommand: updateStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store", "itemId": "item1"},
+ Id:  &id,
  }, nil)
  wg.Wait()
 
@@ -392,12 +392,12 @@ func TestStoreSyncService_UpdateStore(t *testing.T) {
 
  wg.Add(2)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: openStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
+ RequestCommand: openStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
  }, nil)
  bus.SendRequestMessage(syncChan2, &model.Request{
- Request: openStoreRequest,
- Payload: map[string]interface{}{"storeId": "test-store"},
+ RequestCommand: openStoreRequest,
+ Payload:  map[string]interface{}{"storeId": "test-store"},
  }, nil)
  wg.Wait()
 
@@ -407,7 +407,7 @@ func TestStoreSyncService_UpdateStore(t *testing.T) {
  wg.Add(2)
 
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: updateStoreRequest,
+ RequestCommand: updateStoreRequest,
  Payload: map[string]interface{}{
  "storeId": "test-store",
  "itemId": "item3",
@@ -469,7 +469,7 @@ func TestStoreSyncService_UpdateStore(t *testing.T) {
  assert.Equal(t, syncResp1[3], syncResp2[3])
 
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: updateStoreRequest,
+ RequestCommand: updateStoreRequest,
  Payload: map[string]interface{}{
  "storeId": "test-store",
  "itemId": "item4",
@@ -478,7 +478,7 @@ func TestStoreSyncService_UpdateStore(t *testing.T) {
 
  wg.Add(2)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: updateStoreRequest,
+ RequestCommand: updateStoreRequest,
  Payload: map[string]interface{}{
  "storeId": "test-store",
  "itemId": "item3",
@@ -501,7 +501,7 @@ func TestStoreSyncService_UpdateStore(t *testing.T) {
 
  wg.Add(1)
  bus.SendRequestMessage(syncChan, &model.Request{
- Request: updateStoreRequest,
+ RequestCommand: updateStoreRequest,
  Payload: map[string]interface{}{
  "storeId": "test-store",
  "itemId": "item3",

model/message.go

@@ -74,5 +74,7 @@ func decodeResponsePaylod(resp *Response, typ interface{}) error {
  if resp.Error {
  return errors.New(resp.ErrorMessage)
  }
- return mapstructure.Decode(resp.Payload, typ)
+ err := mapstructure.Decode(resp.Payload, typ)
+
+ return err
 }

model/message_test.go

@@ -18,7 +18,7 @@ func TestMessage_CastPayloadToType_HappyPath(t *testing.T) {
 
  // assert
  assert.Nil(t, err)
- assert.EqualValues(t, "dummy-value-coming-through", dest.Request)
+ assert.EqualValues(t, "dummy-value-coming-through", dest.RequestCommand)
 }
 
 func TestMessage_CastPayloadToType_BadPayload(t *testing.T) {
@@ -35,7 +35,7 @@ func TestMessage_CastPayloadToType_BadPayload(t *testing.T) {
  // assert
  assert.NotNil(t, err)
  assert.Contains(t, err.Error(), "failed to unmarshal payload")
- assert.NotEqual(t, "dummy-value-coming-through", dest.Request)
+ assert.NotEqual(t, "dummy-value-coming-through", dest.RequestCommand)
 }
 
 func TestMessage_CastPayloadToType_NonPointer(t *testing.T) {
@@ -48,7 +48,7 @@ func TestMessage_CastPayloadToType_NonPointer(t *testing.T) {
 
  // assert
  assert.NotNil(t, err)
- assert.NotEqual(t, "dummy-value-coming-through", dest.Request)
+ assert.NotEqual(t, "dummy-value-coming-through", dest.RequestCommand)
 }
 
 func TestMessage_CastPayloadToType_NilPointer(t *testing.T) {
@@ -108,7 +108,7 @@ func TestMessage_CastPayloadToType_ErrorResponse(t *testing.T) {
 func getNewTestMessage() *Message {
  rspPayload := &Response{
  Id: &uuid.UUID{},
- Payload: Request{Request: "dummy-value-coming-through"},
+ Payload: Request{RequestCommand: "dummy-value-coming-through"},
  }
 
  jsonEncoded, _ := json.Marshal(rspPayload)

model/request.go

@@ -13,7 +13,7 @@ type Request struct {
  Id *uuid.UUID `json:"id,omitempty"`
  Destination string `json:"channel,omitempty"`
  Payload interface{} `json:"payload,omitempty"`
- RequestCommand string `json:"request,omitempty"`
+ RequestCommand string `json:"request,omitempty" mapstructure:"request"`
  HttpRequest *http.Request `json:"-"`
  HttpResponseWriter http.ResponseWriter `json:"-"`
  // Populated if the request was sent on a "private" channel and