Skip to content
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

Merged
merged 17 commits into from
Aug 25, 2019
Merged

Support handling raw message #138

merged 17 commits into from
Aug 25, 2019

Conversation

rhinof
Copy link
Contributor

@rhinof rhinof commented Aug 23, 2019

Guy Baron and others added 14 commits June 9, 2019 12:01
Merge the 1.0.2 release into master
merge v1.x into master
merge v1.x into master
* add handler metrics to bus and saga (#101)

* add handler metrics to bus and saga + tests

* fix build

* add 0 to the default buckets to catch fast message handling

* PR correction - changed latency to summary(removed bucket configuration), add registration for saga handlers

* PR correction - getting logger as a param

* PR correction - new line in eof

* PR corrections message handler + sync.map + latency as summary

* add rejected messages metric

* dead letter handler should reject messages on failures and rollbacks and ack on commit success  (#105)

* dead letter handler should reject messages on failures and rollbacks

* dead letter handler should reject messages on failures and rollbacks

* dead letter handler should reject messages on failures and rollbacks

* dead letter handler should reject messages on failures and rollbacks

* return an error from the saga store when deleting a saga if saga can not (#110)

be found

In order to deal with concurrent deletes of the sage saga instance we
would wan't to indicate that deleting the saga failed if the saga is not
stored so callers can take proper action

* Persisted timeouts (#107)

* decouple transaction manager from glue

* moved timeout manager to gbus/tx package

* initial commit in order to support persisted timeouts

* first working version of a mysql persisted timeout manager

* fixing ci lint errors

* refactored ensure schema of timeout manager

* cleanup timeout manager when bs shuts down

* fixing formatting issues

* changed logging level from Info to Debug when inserting a new timeout

* resusing timeouts tablename (PR review)

* renamed AcceptTimeoutFunction to SetTimeoutFunction on the
TimeoutManager interface (PR review)

* refactored glue to implement the Logged inetrface and use the GLogged
helper struct

* locking timeout record before executing timeout

In order to prevent having a timeout beeing executed twice due to two
concurrent grabbit instances running the same service a lock (FOR
UPDATE) has been placed on the timeout record  in the scope of the executing transaction

* Commiting the select transaction when querying for pending timeouts

* feat(timeout:lock): This is done in order to reduce contention and allow for parallel processing in case of multiple grabbit instances

* Enable returning a message back from the dead  to the queue (#112)

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* return to q

* return to q

* return to q

* return to q

* return dead to q

* allow no retries

* test - resend dead to queue

* test - resend dead to queue

* test - resend dead to queue

* test - resend dead to queue - fixes after cr

* test - resend dead to queue - fixes after cr

* test - resend dead to queue - fixes after cr

* added metric report on saga timeout (#114)

1) added reporting saga timeouts to the glue component
2) fixed mysql timeoutmanager error when trying to clear a timeout

* Added documentation for grabbit metrics (#117)

* added initial documentation for grabbit metrics

*  including metrics section in readme.md

* fixing goreportcard issues (#118)

* removed logging a warning when worker message channel returns an error (#116)

* corrected saga metrics name and added to metrics documentation (#119)

* corrected saga metrics name and added documentatio

* corrected saga metric name

* corrected typos

* removed non transactional bus mode (#120)
@rhinof rhinof requested review from vladshub, avigailberger and a user August 23, 2019 17:12
gbus/worker.go Show resolved Hide resolved
@vladshub
Copy link
Contributor

Please merge/rebase against v1.x

@@ -30,6 +30,7 @@ type BusConfiguration struct {
type Bus interface {
HandlerRegister
Deadlettering
RawMessageHandling
Copy link
Contributor

@vladshub vladshub Aug 25, 2019

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?

Copy link
Contributor Author

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

        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)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why Global?

Copy link
Contributor Author

Choose a reason for hiding this comment

The 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

*/
type RawMessageHandling interface {
/*
SetGlobalRawMessageHandler registers a handler that gets called for each amqp.Delivery that is delivered
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

please fix indentation :)

@@ -43,7 +43,8 @@ type DefaultBus struct {
amqpOutbox *AMQPOutbox

RPCHandlers map[string]MessageHandler
deadletterHandler DeadLetterMessageHandler
deadletterHandler RawMessageHandler
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Going back to the chain question from before? :)

@coveralls
Copy link

Coverage Status

Coverage increased (+0.3%) to 75.607% when pulling 682ce61 on raw_messages into 58b7cec on v1.x.

@rhinof rhinof merged commit 488c31a into v1.x Aug 25, 2019
@delete-merged-branch delete-merged-branch bot deleted the raw_messages branch August 25, 2019 18:45
rhinof pushed a commit that referenced this pull request Sep 1, 2019
* add handler metrics to bus and saga (#101)

* add handler metrics to bus and saga + tests

* fix build

* add 0 to the default buckets to catch fast message handling

* PR correction - changed latency to summary(removed bucket configuration), add registration for saga handlers

* PR correction - getting logger as a param

* PR correction - new line in eof

* PR corrections message handler + sync.map + latency as summary

* add rejected messages metric

* dead letter handler should reject messages on failures and rollbacks and ack on commit success  (#105)

* dead letter handler should reject messages on failures and rollbacks

* dead letter handler should reject messages on failures and rollbacks

* dead letter handler should reject messages on failures and rollbacks

* dead letter handler should reject messages on failures and rollbacks

* return an error from the saga store when deleting a saga if saga can not (#110)

be found

In order to deal with concurrent deletes of the sage saga instance we
would wan't to indicate that deleting the saga failed if the saga is not
stored so callers can take proper action

* Persisted timeouts (#107)

* decouple transaction manager from glue

* moved timeout manager to gbus/tx package

* initial commit in order to support persisted timeouts

* first working version of a mysql persisted timeout manager

* fixing ci lint errors

* refactored ensure schema of timeout manager

* cleanup timeout manager when bs shuts down

* fixing formatting issues

* changed logging level from Info to Debug when inserting a new timeout

* resusing timeouts tablename (PR review)

* renamed AcceptTimeoutFunction to SetTimeoutFunction on the
TimeoutManager interface (PR review)

* refactored glue to implement the Logged inetrface and use the GLogged
helper struct

* locking timeout record before executing timeout

In order to prevent having a timeout beeing executed twice due to two
concurrent grabbit instances running the same service a lock (FOR
UPDATE) has been placed on the timeout record  in the scope of the executing transaction

* Commiting the select transaction when querying for pending timeouts

* feat(timeout:lock): This is done in order to reduce contention and allow for parallel processing in case of multiple grabbit instances

* Enable returning a message back from the dead  to the queue (#112)

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* return to q

* return to q

* return to q

* return to q

* return dead to q

* allow no retries

* test - resend dead to queue

* test - resend dead to queue

* test - resend dead to queue

* test - resend dead to queue - fixes after cr

* test - resend dead to queue - fixes after cr

* test - resend dead to queue - fixes after cr

* added metric report on saga timeout (#114)

1) added reporting saga timeouts to the glue component
2) fixed mysql timeoutmanager error when trying to clear a timeout

* Added documentation for grabbit metrics (#117)

* added initial documentation for grabbit metrics

*  including metrics section in readme.md

* fixing goreportcard issues (#118)

* removed logging a warning when worker message channel returns an error (#116)

* corrected saga metrics name and added to metrics documentation (#119)

* corrected saga metrics name and added documentatio

* corrected saga metric name

* corrected typos

* removed non transactional bus mode (#120)

* remove fields

* remove fields

* go fmt and go lint error fixes to improve goreportcard (#126)

* go fmt on some files

* go fmt

* added comments on exported types

* cunsume the messages channel via ranging over the channel to prevent (#125)

empty delivreies

* Migrations functionality (#111)

* implement migrations

* implement migrations

* implement migrations

* implement migrations

* implement migrations

* migrations

* migrations

* migrations

* migrations

* migrations

* migrations

* migrations

* fix tests error

* add migrations

* migrations - timeout table migration

* test - resend dead to queue - fixes after cr

* migraration to grabbit (use forked migrator)

* remove fields

* remove fields

* remove fields

* remove fields

* sanitize migrations table name (#130)

* more linting fixes for goreportcard (#129)

* added metrics on deadLetterHandler, refactored HandleDeadLetter inter… (#122)

* added metrics on deadLetterHandler, refactored HandleDeadLetter interface to receive new DeadLetterMessageHandler type

* fix dead letter test and a build error

* added documentation for DeadLetterMessageHandler, also fixed poison spelling throughout code

* retrigger build

* align migrations table name with grabbit convention (#140)

* Improved tracing and added documentation (#142)

* Support handling raw message (#138)

* added call to worker.span.Finish() when exiting processMessage (#145)

* bug fix - when a deadletterhandler panics grabbit fails to reject the… (#136)

* bug fix - when a deadletterhandler panics grabbit fails to reject the message

* bug fix - when a deadletterhandler panics grabbit fails to reject the message

* BPINFRA125 - MERGE MASTER INTO BRANCH

* calling channel.Cancel when worker is stopped (#149)

* Handle empty body messages (#147)

* fixing golint warnings from goreport card (#150)

* more golint fixes (#152)
rhinof pushed a commit that referenced this pull request Sep 19, 2019
* add handler metrics to bus and saga (#101)

* add handler metrics to bus and saga + tests

* fix build

* add 0 to the default buckets to catch fast message handling

* PR correction - changed latency to summary(removed bucket configuration), add registration for saga handlers

* PR correction - getting logger as a param

* PR correction - new line in eof

* PR corrections message handler + sync.map + latency as summary

* add rejected messages metric

* dead letter handler should reject messages on failures and rollbacks and ack on commit success  (#105)

* dead letter handler should reject messages on failures and rollbacks

* dead letter handler should reject messages on failures and rollbacks

* dead letter handler should reject messages on failures and rollbacks

* dead letter handler should reject messages on failures and rollbacks

* return an error from the saga store when deleting a saga if saga can not (#110)

be found

In order to deal with concurrent deletes of the sage saga instance we
would wan't to indicate that deleting the saga failed if the saga is not
stored so callers can take proper action

* Persisted timeouts (#107)

* decouple transaction manager from glue

* moved timeout manager to gbus/tx package

* initial commit in order to support persisted timeouts

* first working version of a mysql persisted timeout manager

* fixing ci lint errors

* refactored ensure schema of timeout manager

* cleanup timeout manager when bs shuts down

* fixing formatting issues

* changed logging level from Info to Debug when inserting a new timeout

* resusing timeouts tablename (PR review)

* renamed AcceptTimeoutFunction to SetTimeoutFunction on the
TimeoutManager interface (PR review)

* refactored glue to implement the Logged inetrface and use the GLogged
helper struct

* locking timeout record before executing timeout

In order to prevent having a timeout beeing executed twice due to two
concurrent grabbit instances running the same service a lock (FOR
UPDATE) has been placed on the timeout record  in the scope of the executing transaction

* Commiting the select transaction when querying for pending timeouts

* feat(timeout:lock): This is done in order to reduce contention and allow for parallel processing in case of multiple grabbit instances

* Enable returning a message back from the dead  to the queue (#112)

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* enable sending raw messages

* return to q

* return to q

* return to q

* return to q

* return dead to q

* allow no retries

* test - resend dead to queue

* test - resend dead to queue

* test - resend dead to queue

* test - resend dead to queue - fixes after cr

* test - resend dead to queue - fixes after cr

* test - resend dead to queue - fixes after cr

* added metric report on saga timeout (#114)

1) added reporting saga timeouts to the glue component
2) fixed mysql timeoutmanager error when trying to clear a timeout

* Added documentation for grabbit metrics (#117)

* added initial documentation for grabbit metrics

*  including metrics section in readme.md

* fixing goreportcard issues (#118)

* removed logging a warning when worker message channel returns an error (#116)

* corrected saga metrics name and added to metrics documentation (#119)

* corrected saga metrics name and added documentatio

* corrected saga metric name

* corrected typos

* removed non transactional bus mode (#120)

* remove fields

* remove fields

* go fmt and go lint error fixes to improve goreportcard (#126)

* go fmt on some files

* go fmt

* added comments on exported types

* cunsume the messages channel via ranging over the channel to prevent (#125)

empty delivreies

* Migrations functionality (#111)

* implement migrations

* implement migrations

* implement migrations

* implement migrations

* implement migrations

* migrations

* migrations

* migrations

* migrations

* migrations

* migrations

* migrations

* fix tests error

* add migrations

* migrations - timeout table migration

* test - resend dead to queue - fixes after cr

* migraration to grabbit (use forked migrator)

* remove fields

* remove fields

* remove fields

* remove fields

* sanitize migrations table name (#130)

* more linting fixes for goreportcard (#129)

* added metrics on deadLetterHandler, refactored HandleDeadLetter inter… (#122)

* added metrics on deadLetterHandler, refactored HandleDeadLetter interface to receive new DeadLetterMessageHandler type

* fix dead letter test and a build error

* added documentation for DeadLetterMessageHandler, also fixed poison spelling throughout code

* retrigger build

* align migrations table name with grabbit convention (#140)

* Improved tracing and added documentation (#142)

* Support handling raw message (#138)

* added call to worker.span.Finish() when exiting processMessage (#145)

* bug fix - when a deadletterhandler panics grabbit fails to reject the… (#136)

* bug fix - when a deadletterhandler panics grabbit fails to reject the message

* bug fix - when a deadletterhandler panics grabbit fails to reject the message

* BPINFRA125 - MERGE MASTER INTO BRANCH

* calling channel.Cancel when worker is stopped (#149)

* Handle empty body messages (#147)

* fixing golint warnings from goreport card (#150)

* more golint fixes (#152)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

None yet

3 participants