-
Notifications
You must be signed in to change notification settings - Fork 335
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
### Motivation Add A chain of interceptors for Producer and Consumer as an option, these interceptors will be called at some points, it can be used for tracing, metrics, and so on. ### Modifications Add two files for interceptor definition. Call interceptor's methods at appropriate position. * review test Co-authored-by: lijingfeng <lijingfeng@laiye.com>
- Loading branch information
1 parent
5fea0e8
commit b434511
Showing
11 changed files
with
382 additions
and
2 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,51 @@ | ||
// Licensed to the Apache Software Foundation (ASF) under one | ||
// or more contributor license agreements. See the NOTICE file | ||
// distributed with this work for additional information | ||
// regarding copyright ownership. The ASF licenses this file | ||
// to you under the Apache License, Version 2.0 (the | ||
// "License"); you may not use this file except in compliance | ||
// with the License. You may obtain a copy of the License at | ||
// | ||
// http://www.apache.org/licenses/LICENSE-2.0 | ||
// | ||
// Unless required by applicable law or agreed to in writing, | ||
// software distributed under the License is distributed on an | ||
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
// KIND, either express or implied. See the License for the | ||
// specific language governing permissions and limitations | ||
// under the License. | ||
|
||
package pulsar | ||
|
||
type ConsumerInterceptor interface { | ||
// BeforeConsume This is called just before the message is send to Consumer's ConsumerMessage channel. | ||
BeforeConsume(message ConsumerMessage) | ||
|
||
// OnAcknowledge This is called consumer sends the acknowledgment to the broker. | ||
OnAcknowledge(consumer Consumer, msgID MessageID) | ||
|
||
// OnNegativeAcksSend This method will be called when a redelivery from a negative acknowledge occurs. | ||
OnNegativeAcksSend(consumer Consumer, msgIDs []MessageID) | ||
} | ||
|
||
type ConsumerInterceptors []ConsumerInterceptor | ||
|
||
func (x ConsumerInterceptors) BeforeConsume(message ConsumerMessage) { | ||
for i := range x { | ||
x[i].BeforeConsume(message) | ||
} | ||
} | ||
|
||
func (x ConsumerInterceptors) OnAcknowledge(consumer Consumer, msgID MessageID) { | ||
for i := range x { | ||
x[i].OnAcknowledge(consumer, msgID) | ||
} | ||
} | ||
|
||
func (x ConsumerInterceptors) OnNegativeAcksSend(consumer Consumer, msgIDs []MessageID) { | ||
for i := range x { | ||
x[i].OnNegativeAcksSend(consumer, msgIDs) | ||
} | ||
} | ||
|
||
var defaultConsumerInterceptors = make(ConsumerInterceptors, 0) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,43 @@ | ||
// Licensed to the Apache Software Foundation (ASF) under one | ||
// or more contributor license agreements. See the NOTICE file | ||
// distributed with this work for additional information | ||
// regarding copyright ownership. The ASF licenses this file | ||
// to you under the Apache License, Version 2.0 (the | ||
// "License"); you may not use this file except in compliance | ||
// with the License. You may obtain a copy of the License at | ||
// | ||
// http://www.apache.org/licenses/LICENSE-2.0 | ||
// | ||
// Unless required by applicable law or agreed to in writing, | ||
// software distributed under the License is distributed on an | ||
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
// KIND, either express or implied. See the License for the | ||
// specific language governing permissions and limitations | ||
// under the License. | ||
|
||
package pulsar | ||
|
||
type ProducerInterceptor interface { | ||
// BeforeSend This is called before send the message to the brokers. This method is allowed to modify the | ||
BeforeSend(producer Producer, message *ProducerMessage) | ||
|
||
// OnSendAcknowledgement This method is called when the message sent to the broker has been acknowledged, | ||
// or when sending the message fails. | ||
OnSendAcknowledgement(producer Producer, message *ProducerMessage, msgID MessageID) | ||
} | ||
|
||
type ProducerInterceptors []ProducerInterceptor | ||
|
||
func (x ProducerInterceptors) BeforeSend(producer Producer, message *ProducerMessage) { | ||
for i := range x { | ||
x[i].BeforeSend(producer, message) | ||
} | ||
} | ||
|
||
func (x ProducerInterceptors) OnSendAcknowledgement(producer Producer, message *ProducerMessage, msgID MessageID) { | ||
for i := range x { | ||
x[i].OnSendAcknowledgement(producer, message, msgID) | ||
} | ||
} | ||
|
||
var defaultProducerInterceptors = make(ProducerInterceptors, 0) |
Oops, something went wrong.