diff --git a/application-framework/event/eventService.go b/application-framework/event/eventService.go index f51bb086e989073f46ed06dab537d26f0266576a..930a848d24ca4cfaecd796a1c833b5cfc275f526 100644 --- a/application-framework/event/eventService.go +++ b/application-framework/event/eventService.go @@ -87,7 +87,7 @@ func (e *Service) setupQueueConsume(topic string, stopChan chan os.Signal) error nil, // arguments ) if err != nil { - return &customerrs.AMQPMessageFailError{Action: "failed declaring queue", Err: err} + return &customerrs.AMQPInitFailError{Action: "failed declaring exchange", Err: err} } q, err := e.channel.QueueDeclare( @@ -99,7 +99,7 @@ func (e *Service) setupQueueConsume(topic string, stopChan chan os.Signal) error nil, // arguments ) if err != nil { - return &customerrs.AMQPMessageFailError{Action: "failed declaring queue", Err: err} + return &customerrs.AMQPInitFailError{Action: "failed declaring queue", Err: err} } err = e.channel.QueueBind( @@ -110,7 +110,7 @@ func (e *Service) setupQueueConsume(topic string, stopChan chan os.Signal) error nil, ) if err != nil { - return &customerrs.AMQPMessageFailError{Action: "failed declaring queue", Err: err} + return &customerrs.AMQPInitFailError{Action: "failed binding to queue to exchange", Err: err} } msgs, err := e.channel.Consume( diff --git a/controller/eventService/Service.go b/controller/eventService/Service.go index e3429d78ad7920de8b68d112fd268c959f786927..a3041b8bdc01745ba3203852d0fdd031f4786f82 100644 --- a/controller/eventService/Service.go +++ b/controller/eventService/Service.go @@ -180,7 +180,7 @@ func (e *EventService) PublishEvent(topic string, event event.Event) error { nil, // arguments ) if err != nil { - return &customerrs.AMQPInitFailError{Action: "failed declaring queue", Err: err} + return &customerrs.AMQPInitFailError{Action: "failed declaring exchange", Err: err} } eventBody, err := json.Marshal(event) @@ -188,8 +188,11 @@ func (e *EventService) PublishEvent(topic string, event event.Event) error { return &customerrs.CouldNotMarshallError{Identifier: topic + " " + event.EntityID.String(), Type: event.Type, Err: err} } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + err = e.channel.PublishWithContext( - context.TODO(), + ctx, topic, // exchange "", // routing key false, // mandatory