-
Notifications
You must be signed in to change notification settings - Fork 20
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Support handling raw message #138
Changes from 14 commits
51a7b20
5624310
5d077c4
7a9f639
44f71ca
31d9fc5
4504725
78dfa6e
7a474ea
c28c519
0df769c
a683f28
2787d02
e061bd9
a916269
b4d07df
682ce61
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -30,6 +30,7 @@ type BusConfiguration struct { | |
type Bus interface { | ||
HandlerRegister | ||
Deadlettering | ||
RawMessageHandling | ||
BusSwitch | ||
Messaging | ||
SagaRegister | ||
|
@@ -129,10 +130,37 @@ type Saga interface { | |
|
||
//Deadlettering provides the ability to handle messages that were rejected as poision and arrive to the deadletter queue | ||
type Deadlettering interface { | ||
HandleDeadletter(handler DeadLetterMessageHandler) | ||
/* | ||
HandleDeadletter is deprecated use RawMessageHandling.SetGlobalRawMessageHandler instead. | ||
This function will be removed in future grabbit releases | ||
*/ | ||
HandleDeadletter(handler RawMessageHandler) | ||
ReturnDeadToQueue(ctx context.Context, publishing *amqp.Publishing) error | ||
} | ||
|
||
/* | ||
RawMessageHandling provides the ability to consume and send raq amqp messages with the transactional guarantees | ||
that the bus provides | ||
*/ | ||
type RawMessageHandling interface { | ||
/* | ||
SetGlobalRawMessageHandler registers a handler that gets called for each amqp.Delivery that is delivered | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. please fix indentation :) |
||
to the service queue. | ||
The handler will get called with a scoped transaction that is a different transaction than the ones that | ||
regular message handlers are scoped by as we want the RawMessage handler to get executed even if the amqp.Delivery | ||
can not be serialized by the bus to one of the registered schemas | ||
|
||
In case a bus has both a raw message handler and regular ones the bus will first call the raw message handler | ||
and afterward will call any registered message handlers. | ||
if the global raw handler returns an error the message gets rejected and any additional | ||
handlers will not be called. | ||
You should not use the global raw message handler to drive business logic as it breaks the local transactivity | ||
guarantees grabbit provides and should only be used in specialized cases. | ||
If you do decide to use this feature try not shooting yourself in the foot. | ||
*/ | ||
SetGlobalRawMessageHandler(handler RawMessageHandler) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Why Global? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I do not see the case in which more than one is needed. and if a client would like some logic being routed to according to specific criteria they should use the regular message handlers |
||
} | ||
|
||
//RequestSagaTimeout is the interface a saga needs to implement to get timeout servicess | ||
type RequestSagaTimeout interface { | ||
TimeoutDuration() time.Duration | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -43,7 +43,8 @@ type DefaultBus struct { | |
amqpOutbox *AMQPOutbox | ||
|
||
RPCHandlers map[string]MessageHandler | ||
deadletterHandler DeadLetterMessageHandler | ||
deadletterHandler RawMessageHandler | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Going back to the chain question from before? :) |
||
globalRawHandler RawMessageHandler | ||
HandlersLock *sync.Mutex | ||
RPCLock *sync.Mutex | ||
SenderLock *sync.Mutex | ||
|
@@ -73,8 +74,8 @@ var ( | |
//BaseRetryDuration defines the basic milliseconds that the retry algorithm uses | ||
//for a random retry time. Default is 10 but it is configurable. | ||
BaseRetryDuration = 10 * time.Millisecond | ||
//RpcHeaderName used to define the header in grabbit for RPC | ||
RpcHeaderName = "x-grabbit-msg-rpc-id" | ||
//RPCHeaderName used to define the header in grabbit for RPC | ||
RPCHeaderName = "x-grabbit-msg-rpc-id" | ||
) | ||
|
||
func (b *DefaultBus) createRPCQueue() (amqp.Queue, error) { | ||
|
@@ -286,6 +287,7 @@ func (b *DefaultBus) createBusWorkers(workerNum uint) ([]*worker, error) { | |
rpcLock: b.RPCLock, | ||
rpcHandlers: b.RPCHandlers, | ||
deadletterHandler: b.deadletterHandler, | ||
globalRawHandler: b.globalRawHandler, | ||
handlersLock: &sync.Mutex{}, | ||
registrations: b.Registrations, | ||
serializer: b.Serializer, | ||
|
@@ -547,11 +549,17 @@ func (b *DefaultBus) HandleEvent(exchange, topic string, event Message, handler | |
return b.registerHandlerImpl(exchange, topic, event, handler) | ||
} | ||
|
||
//HandleDeadletter implements GBus.HandleDeadletter | ||
func (b *DefaultBus) HandleDeadletter(handler DeadLetterMessageHandler) { | ||
//HandleDeadletter implements Deadlettering.HandleDeadletter | ||
func (b *DefaultBus) HandleDeadletter(handler RawMessageHandler) { | ||
b.registerDeadLetterHandler(handler) | ||
} | ||
|
||
//HandleDeadletter implements RawMessageHandling.SetGlobalRawMessageHandler | ||
func (b *DefaultBus) SetGlobalRawMessageHandler(handler RawMessageHandler) { | ||
metrics.AddHandlerMetrics(handler.Name()) | ||
b.globalRawHandler = handler | ||
} | ||
|
||
//ReturnDeadToQueue returns a message to its original destination | ||
func (b *DefaultBus) ReturnDeadToQueue(ctx context.Context, publishing *amqp.Publishing) error { | ||
return b.returnDeadToQueue(ctx, nil, publishing) | ||
|
@@ -691,7 +699,7 @@ func (b *DefaultBus) registerHandlerImpl(exchange, routingKey string, msg Messag | |
return nil | ||
} | ||
|
||
func (b *DefaultBus) registerDeadLetterHandler(handler DeadLetterMessageHandler) { | ||
func (b *DefaultBus) registerDeadLetterHandler(handler RawMessageHandler) { | ||
metrics.AddHandlerMetrics(handler.Name()) | ||
b.deadletterHandler = handler | ||
} | ||
|
@@ -705,7 +713,7 @@ type rpcPolicy struct { | |
} | ||
|
||
func (p rpcPolicy) Apply(publishing *amqp.Publishing) { | ||
publishing.Headers[RpcHeaderName] = p.rpcID | ||
publishing.Headers[RPCHeaderName] = p.rpcID | ||
} | ||
|
||
//Log returns the default logrus.FieldLogger for the bus via the Glogged helper | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Wouldn't a chain design with default chain components would be more effective here and more clear?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Perhaps, we can explore this design idea