You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2023/01/06 14:10:14 UTC
[GitHub] [pulsar-client-cpp] RobertIndie opened a new pull request, #169: [feat] Add producer interceptor
RobertIndie opened a new pull request, #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169
<!--
### Contribution Checklist
- PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://docs.google.com/document/d/1d8Pw6ZbWk-_pCKdOmdvx9rnhPiyuxwq60_TrD68d7BA/edit#heading=h.trs9rsex3xom)*.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
-->
Master Issue: #150
### Motivation
This is the producer interceptor implementation of #150.
### Modifications
* Add ProducerInterceptor interface
* Triggering the interceptor when `beforeSend`, `onSendAcknowledgement`, `onPartitionsChange` and `close`.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
This change added tests.
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [x] `doc-required`
(Your PR needs to update docs and you will update later)
- [ ] `doc-not-needed`
(Please explain why)
- [ ] `doc`
(Your PR contains doc changes)
- [ ] `doc-complete`
(Docs have been already added)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on pull request #169: [feat] Add producer interceptor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#issuecomment-1432626323
Could you solve the conflicts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1104343135
##########
tests/InterceptorsTest.cc:
##########
@@ -0,0 +1,157 @@
+/**
+ * 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.
+ */
+#include <gtest/gtest.h>
+#include <pulsar/Client.h>
+#include <pulsar/ProducerInterceptor.h>
+
+#include <utility>
+
+#include "HttpHelper.h"
+#include "Latch.h"
+
+static const std::string serviceUrl = "pulsar://localhost:6650";
+static const std::string adminUrl = "http://localhost:8080/";
+
+using namespace pulsar;
+
+class TestInterceptor : public ProducerInterceptor {
+ public:
+ explicit TestInterceptor(Latch latch) : latch_(std::move(latch)) {}
+
+ Message beforeSend(const Producer& producer, const Message& message) override {
+ return MessageBuilder().setProperty("key", "set").setContent(message.getDataAsString()).build();
+ }
+
+ void onSendAcknowledgement(const Producer& producer, Result result, const Message& message,
+ const MessageId& messageID) override {
+ ASSERT_EQ(result, ResultOk);
+ auto properties = message.getProperties();
+ if (properties.find("key") != properties.end() && properties["key"] == "set") {
+ latch_.countdown();
+ }
+ }
+
+ private:
+ Latch latch_;
+};
+
+class ExceptionInterceptor : public ProducerInterceptor {
+ public:
+ explicit ExceptionInterceptor(Latch latch) : latch_(std::move(latch)) {}
+
+ Message beforeSend(const Producer& producer, const Message& message) override {
+ latch_.countdown();
+ throw std::runtime_error("expected exception");
+ }
+
+ void onSendAcknowledgement(const Producer& producer, Result result, const Message& message,
+ const MessageId& messageID) override {
+ latch_.countdown();
+ throw std::runtime_error("expected exception");
+ }
+
+ private:
+ Latch latch_;
+};
+
+class PartitionsChangeInterceptor : public ProducerInterceptor {
+ public:
+ explicit PartitionsChangeInterceptor(Latch latch) : latch_(std::move(latch)) {}
+
+ Message beforeSend(const Producer& producer, const Message& message) override { return message; }
+
+ void onSendAcknowledgement(const Producer& producer, Result result, const Message& message,
+ const MessageId& messageID) override {}
+
+ void onPartitionsChange(const std::string& topicName, int partitions) override {
+ ASSERT_EQ(partitions, 3);
+ latch_.countdown();
+ }
+
+ private:
+ Latch latch_;
+};
+
+TEST(InterceptorsTest, testProducerInterceptor) {
+ const std::string topic = "InterceptorsTest-testProducerInterceptor-" + std::to_string(time(nullptr));
+
+ Latch latch(1);
+
+ Client client(serviceUrl);
+ ProducerConfiguration conf;
+ conf.intercept({std::make_shared<TestInterceptor>(latch)});
+ Producer producer;
+ client.createProducer(topic, conf, producer);
+
+ Message msg = MessageBuilder().setContent("content").build();
+ Result result = producer.send(msg);
+ ASSERT_EQ(result, ResultOk);
+
+ latch.wait(std::chrono::seconds(5));
Review Comment:
Thanks. I have added the check for the return value
> as well as the received message, which should have set the "key" property through the interceptor.
Already did that in the `TestInterceptor.onSendAcknowledgement()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] Anonymitaet commented on pull request #169: [feat] Add producer interceptor
Posted by "Anonymitaet (via GitHub)" <gi...@apache.org>.
Anonymitaet commented on PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#issuecomment-1510855088
Hi @RobertIndie, thanks for introducing this feature!
As we discussed just now, for the doc side:
### ✅ Pulsar feature matrix
I've ticked the boxes of ["Interceptors" for C++ producer and consumer\(https://docs.google.com/spreadsheets/d/1YHYTkIXR8-Ql103u-IMI18TXLlGStK8uJjDsOOA0T20/edit#gid=1784579914&range=48:48)
### ✅ Pulsar client docs
Could you please provide the following tech inputs? TIA!
The following docs are only added to `NEXT` since this feature is server-independent.
- [C++ > Producer > Interceptor](https://pulsar.apache.org/docs/next/client-libraries-producers/#intercept-messages)
<img width="1919" alt="image" src="https://user-images.githubusercontent.com/50226895/232416133-a1c4c9c5-1127-4600-9570-5084ad088166.png">
- [C++ > Consumer > Interceptor](https://pulsar.apache.org/docs/next/client-libraries-consumers/#intercept-messages)
<img width="1919" alt="image" src="https://user-images.githubusercontent.com/50226895/232416961-e3970742-c4ce-44bf-ba2f-00d2a4f5e96f.png">
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1109613473
##########
lib/ProducerInterceptors.h:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.
+ */
+
+#pragma once
+
+#include <pulsar/ProducerInterceptor.h>
+
+#include <atomic>
+#include <utility>
+#include <vector>
+
+namespace pulsar {
+
+class ProducerInterceptors {
+ public:
+ explicit ProducerInterceptors(std::vector<ProducerInterceptorPtr> interceptors)
+ : interceptors_(std::move(interceptors)) {}
Review Comment:
Even if you use `std::move(interceptors)`, the copy will still happen because this constructor accepts an argument by value, not reference.
For example,
```c++
#include <iostream>
#include <vector>
using namespace std;
struct Foo {
Foo() { cout << "Foo()" << endl; }
Foo(const Foo &) { cout << "Foo copy" << endl; }
Foo(Foo &&foo) noexcept { cout << "Foo move" << endl; }
};
struct FooVector {
FooVector(vector<Foo> v) : v_(std::move(v)) {}
vector<Foo> v_;
};
int main(int argc, char* argv[]) {
vector<Foo> v;
v.push_back(Foo{});
cout << "--------" << endl;
FooVector foos{v};
return 0;
}
```
You will see the following output:
```
Foo()
Foo move
--------
Foo copy
```
If you want to avoid the copy, you have to pass `std::move(v)` like:
```c++
FooVector foos{std::move(v)};
```
However, in your case, you call it here:
```c++
auto interceptors = std::make_shared<ProducerInterceptors>(conf.getInterceptors());
```
If you changed it to `std::move(conf.getInterceptors())`, the `conf` will hold no interceptor and an unexpected behavior might happen (imagine users might retrieve the interceptors from the `ProducerConfiguration` instance).
Therefore, in short, it's better just to keep the copy semantics instead of the move semantics:
```c++
explicit ProducerInterceptors(const std::vector<ProducerInterceptorPtr>& interceptors) : interceptors_(interceptors) {}
```
This is not a critical path, so copying is acceptable here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on pull request #169: [feat] Add producer interceptor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#issuecomment-1436486228
@RobertIndie The main branch is broken due to a recent upgrade of `macos-12` runner image. The fix is here: https://github.com/apache/pulsar-client-cpp/pull/199
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1065684370
##########
include/pulsar/ProducerInterceptor.h:
##########
@@ -0,0 +1,114 @@
+/**
+ * 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.
+ */
+
+#ifndef PULSAR_PRODUCER_INTERCEPTOR_H
+#define PULSAR_PRODUCER_INTERCEPTOR_H
+
+#include <pulsar/Message.h>
+#include <pulsar/Producer.h>
+#include <pulsar/defines.h>
+
+/**
+ * An interface that allows you to intercept (and possibly mutate) the
+ * messages received by the producer before they are published to the Pulsar
+ * brokers.
+ *
+ * <p>Exceptions thrown by ProducerInterceptor methods will be caught, logged, but
+ * not propagated further.
+ *
+ * <p>ProducerInterceptor callbacks may be called from multiple threads. Interceptor
+ * implementation must ensure thread-safety, if needed.
+ */
+namespace pulsar {
+
+class Producer;
+
+class PULSAR_PUBLIC ProducerInterceptor {
+ public:
+ virtual ~ProducerInterceptor() {}
+
+ /**
+ * Close the interceptor
+ */
+ virtual void close() {}
+
+ /**
+ * This is called from Producer#send and Producer#sendAsync methods, before
+ * send the message to the brokers. This method is allowed to modify the
+ * record, in which case, the new record will be returned.
+ *
+ * <p>Any exception thrown by this method will be caught by the caller and
+ * logged, but not propagated further.
+ *
+ * <p>Since the producer may run multiple interceptors, a particular
+ * interceptor's #beforeSend(Producer, Message) callback will be called in the
+ * order specified by ProducerConfiguration#intercept().
+ *
+ * <p>The first interceptor in the list gets the message passed from the client,
+ * the following interceptor will be passed the message returned by the
+ * previous interceptor, and so on. Since interceptors are allowed to modify
+ * messages, interceptors may potentially get the message already modified by
+ * other interceptors. However, building a pipeline of mutable interceptors
+ * that depend on the output of the previous interceptor is discouraged,
+ * because of potential side-effects caused by interceptors potentially
+ * failing to modify the message and throwing an exception. If one of the
+ * interceptors in the list throws an exception from beforeSend(Message),
+ * the exception is caught, logged, and the next interceptor is called with
+ * the message returned by the last successful interceptor in the list,
+ * or otherwise the client.
+ *
+ * @param producer the producer which contains the interceptor.
+ * @param message message to send.
+ * @return the intercepted message.
+ */
+ virtual Message beforeSend(const Producer& producer, const Message& message) = 0;
+
+ /**
+ * This method is called when the message sent to the broker has been
+ * acknowledged, or when sending the message fails.
+ * This method is generally called just before the user callback is
+ * called.
+ *
+ * <p>Any exception thrown by this method will be ignored by the caller.
+ *
+ * <p>This method will generally execute in the background I/O thread, so the
+ * implementation should be reasonably fast. Otherwise, sending of messages
+ * from other threads could be delayed.
+ *
+ * @param producer the producer which contains the interceptor.
+ * @param result the result for sending messages, ResultOk indicates send has succeed.
+ * @param message the message that application sends.
+ * @param messageID the message id that assigned by the broker.
+ */
+ virtual void onSendAcknowledgement(const Producer& producer, Result result, const Message& message,
+ const MessageId& messageID) = 0;
+
+ /**
+ * This method is called when partitions of the topic (partitioned-topic) changes.
+ *
+ * @param topicName topic name
+ * @param partitions new updated partitions
+ */
+ virtual void onPartitionsChange(const std::string& topicName, const int partitions) {}
Review Comment:
```suggestion
virtual void onPartitionsChange(const std::string& topicName, int partitions) {}
```
Don't add `const` to value types in function parameters. See https://stackoverflow.com/questions/117293/use-of-const-for-function-parameters
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1066677139
##########
include/pulsar/Producer.h:
##########
@@ -23,13 +23,16 @@
#include <pulsar/defines.h>
#include <stdint.h>
+#include <functional>
#include <memory>
namespace pulsar {
class ProducerImplBase;
class PulsarWrapper;
class PulsarFriend;
+typedef std::function<void(Result, const MessageId& messageId)> SendCallback;
+typedef std::function<void(Result)> CloseCallback;
Review Comment:
Why did you add these extra definitions?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on pull request #169: [feat] Add producer interceptor
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#issuecomment-1432860112
> Could you solve the conflicts?
Done. PTAL again. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1104343765
##########
include/pulsar/Producer.h:
##########
@@ -23,13 +23,16 @@
#include <pulsar/defines.h>
#include <stdint.h>
+#include <functional>
#include <memory>
namespace pulsar {
class ProducerImplBase;
class PulsarWrapper;
class PulsarFriend;
+typedef std::function<void(Result, const MessageId& messageId)> SendCallback;
+typedef std::function<void(Result)> CloseCallback;
Review Comment:
Fixed by https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1066689463
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1071937413
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
Review Comment:
I just think of a point that if users want to maintain the interceptors, e.g. there is already a `std::vector<ProducerInterceptorPtr>`, there is no way to convert a `vector` to `initializer_list`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1104344147
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
Review Comment:
OK. I have decided to use `vector` here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1111581516
##########
lib/ProducerInterceptors.h:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.
+ */
+
+#pragma once
+
+#include <pulsar/ProducerInterceptor.h>
+
+#include <atomic>
+#include <utility>
+#include <vector>
+
+namespace pulsar {
+
+class ProducerInterceptors {
+ public:
+ explicit ProducerInterceptors(std::vector<ProducerInterceptorPtr> interceptors)
+ : interceptors_(std::move(interceptors)) {}
Review Comment:
BTW, I'm also okay with the current method signature.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie merged pull request #169: [feat] Add producer interceptor
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie merged PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1104344875
##########
include/pulsar/ProducerInterceptor.h:
##########
@@ -0,0 +1,114 @@
+/**
+ * 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.
+ */
+
+#ifndef PULSAR_PRODUCER_INTERCEPTOR_H
+#define PULSAR_PRODUCER_INTERCEPTOR_H
+
+#include <pulsar/Message.h>
+#include <pulsar/Producer.h>
+#include <pulsar/defines.h>
+
+/**
+ * An interface that allows you to intercept (and possibly mutate) the
+ * messages received by the producer before they are published to the Pulsar
+ * brokers.
+ *
+ * <p>Exceptions thrown by ProducerInterceptor methods will be caught, logged, but
+ * not propagated further.
+ *
+ * <p>ProducerInterceptor callbacks may be called from multiple threads. Interceptor
+ * implementation must ensure thread-safety, if needed.
+ */
+namespace pulsar {
+
+class Producer;
+
+class PULSAR_PUBLIC ProducerInterceptor {
+ public:
+ virtual ~ProducerInterceptor() {}
+
+ /**
+ * Close the interceptor
+ */
+ virtual void close() {}
Review Comment:
Yes. Fixed it. The java client also need to be fixed.
##########
include/pulsar/ProducerInterceptor.h:
##########
@@ -0,0 +1,114 @@
+/**
+ * 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.
+ */
+
+#ifndef PULSAR_PRODUCER_INTERCEPTOR_H
+#define PULSAR_PRODUCER_INTERCEPTOR_H
+
+#include <pulsar/Message.h>
+#include <pulsar/Producer.h>
Review Comment:
Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1066673285
##########
tests/InterceptorsTest.cc:
##########
@@ -0,0 +1,157 @@
+/**
+ * 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.
+ */
+#include <gtest/gtest.h>
+#include <pulsar/Client.h>
+#include <pulsar/ProducerInterceptor.h>
+
+#include <utility>
+
+#include "HttpHelper.h"
+#include "Latch.h"
+
+static const std::string serviceUrl = "pulsar://localhost:6650";
+static const std::string adminUrl = "http://localhost:8080/";
+
+using namespace pulsar;
+
+class TestInterceptor : public ProducerInterceptor {
+ public:
+ explicit TestInterceptor(Latch latch) : latch_(std::move(latch)) {}
+
+ Message beforeSend(const Producer& producer, const Message& message) override {
+ return MessageBuilder().setProperty("key", "set").setContent(message.getDataAsString()).build();
+ }
+
+ void onSendAcknowledgement(const Producer& producer, Result result, const Message& message,
+ const MessageId& messageID) override {
+ ASSERT_EQ(result, ResultOk);
+ auto properties = message.getProperties();
+ if (properties.find("key") != properties.end() && properties["key"] == "set") {
+ latch_.countdown();
+ }
+ }
+
+ private:
+ Latch latch_;
+};
+
+class ExceptionInterceptor : public ProducerInterceptor {
+ public:
+ explicit ExceptionInterceptor(Latch latch) : latch_(std::move(latch)) {}
+
+ Message beforeSend(const Producer& producer, const Message& message) override {
+ latch_.countdown();
+ throw std::runtime_error("expected exception");
+ }
+
+ void onSendAcknowledgement(const Producer& producer, Result result, const Message& message,
+ const MessageId& messageID) override {
+ latch_.countdown();
+ throw std::runtime_error("expected exception");
+ }
+
+ private:
+ Latch latch_;
+};
+
+class PartitionsChangeInterceptor : public ProducerInterceptor {
+ public:
+ explicit PartitionsChangeInterceptor(Latch latch) : latch_(std::move(latch)) {}
+
+ Message beforeSend(const Producer& producer, const Message& message) override { return message; }
+
+ void onSendAcknowledgement(const Producer& producer, Result result, const Message& message,
+ const MessageId& messageID) override {}
+
+ void onPartitionsChange(const std::string& topicName, int partitions) override {
+ ASSERT_EQ(partitions, 3);
+ latch_.countdown();
+ }
+
+ private:
+ Latch latch_;
+};
+
+TEST(InterceptorsTest, testProducerInterceptor) {
+ const std::string topic = "InterceptorsTest-testProducerInterceptor-" + std::to_string(time(nullptr));
+
+ Latch latch(1);
+
+ Client client(serviceUrl);
+ ProducerConfiguration conf;
+ conf.intercept({std::make_shared<TestInterceptor>(latch)});
+ Producer producer;
+ client.createProducer(topic, conf, producer);
+
+ Message msg = MessageBuilder().setContent("content").build();
+ Result result = producer.send(msg);
+ ASSERT_EQ(result, ResultOk);
+
+ latch.wait(std::chrono::seconds(5));
Review Comment:
You should check the returned value of `wait`, as well as the received message, which should have set the "key" property through the interceptor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1065690028
##########
lib/ProducerImpl.cc:
##########
@@ -975,5 +984,33 @@ void ProducerImpl::asyncWaitSendTimeout(DurationType expiryTime) {
ProducerImplWeakPtr ProducerImpl::weak_from_this() noexcept { return shared_from_this(); }
+Message ProducerImpl::beforeSend(const Producer& producer, const Message& message) const {
+ if (interceptors_.size() == 0) {
Review Comment:
```suggestion
if (interceptors_.empty()) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] shibd commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
shibd commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1066668816
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
Review Comment:
Is it better to pass a reference here?
##########
include/pulsar/Producer.h:
##########
@@ -23,13 +23,16 @@
#include <pulsar/defines.h>
#include <stdint.h>
+#include <functional>
#include <memory>
namespace pulsar {
class ProducerImplBase;
class PulsarWrapper;
class PulsarFriend;
+typedef std::function<void(Result, const MessageId& messageId)> SendCallback;
Review Comment:
You don't need to redefine these here. Remove the import of `Producer.h` from `ProducerInterceptor.h` and the compilation will succeed
##########
include/pulsar/ProducerInterceptor.h:
##########
@@ -0,0 +1,114 @@
+/**
+ * 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.
+ */
+
+#ifndef PULSAR_PRODUCER_INTERCEPTOR_H
+#define PULSAR_PRODUCER_INTERCEPTOR_H
+
+#include <pulsar/Message.h>
+#include <pulsar/Producer.h>
Review Comment:
Line 40 `Producer` has been declared. There is no need to import `Producer.h` here.
##########
include/pulsar/ProducerInterceptor.h:
##########
@@ -0,0 +1,114 @@
+/**
+ * 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.
+ */
+
+#ifndef PULSAR_PRODUCER_INTERCEPTOR_H
+#define PULSAR_PRODUCER_INTERCEPTOR_H
+
+#include <pulsar/Message.h>
+#include <pulsar/Producer.h>
+#include <pulsar/defines.h>
+
+/**
+ * An interface that allows you to intercept (and possibly mutate) the
+ * messages received by the producer before they are published to the Pulsar
+ * brokers.
+ *
+ * <p>Exceptions thrown by ProducerInterceptor methods will be caught, logged, but
+ * not propagated further.
+ *
+ * <p>ProducerInterceptor callbacks may be called from multiple threads. Interceptor
+ * implementation must ensure thread-safety, if needed.
+ */
+namespace pulsar {
+
+class Producer;
+
+class PULSAR_PUBLIC ProducerInterceptor {
+ public:
+ virtual ~ProducerInterceptor() {}
+
+ /**
+ * Close the interceptor
+ */
+ virtual void close() {}
Review Comment:
When the producer close, do we need to invoke this close()? I see this is not invoked in Java Client, maybe it a bug?
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
+
+ std::vector<ProducerInterceptorPtr> getInterceptors() const;
Review Comment:
return `const std::vector<ProducerInterceptorPtr>&`?
##########
include/pulsar/ProducerInterceptor.h:
##########
@@ -0,0 +1,114 @@
+/**
+ * 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.
+ */
+
+#ifndef PULSAR_PRODUCER_INTERCEPTOR_H
+#define PULSAR_PRODUCER_INTERCEPTOR_H
+
+#include <pulsar/Message.h>
+#include <pulsar/Producer.h>
Review Comment:
ProducerInterceptor.h -> Producer.h -> ProducerConfiguration.h -> ProducerInterceptor.h
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
RobertIndie commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1069152267
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
Review Comment:
I don't have a strong opinion on it. Just because `initializer_list ` is more often used as the lightweight container type for the input parameter. I could change to using vector anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1111557113
##########
lib/ProducerInterceptors.h:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.
+ */
+
+#pragma once
+
+#include <pulsar/ProducerInterceptor.h>
+
+#include <atomic>
+#include <utility>
+#include <vector>
+
+namespace pulsar {
+
+class ProducerInterceptors {
+ public:
+ explicit ProducerInterceptors(std::vector<ProducerInterceptorPtr> interceptors)
+ : interceptors_(std::move(interceptors)) {}
Review Comment:
It's a warning from the clang-tidy: `Clang-Tidy: Parameter 'interceptors' is passed by value and only copied once; consider moving it to avoid unnecessary copies `.
We should copy the value here and it's okay to do that because it's a pointer type. It will raise a warning here if we don't use `std::move`. But I'm okay for both.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1109631821
##########
lib/ProducerImplBase.h:
##########
@@ -48,6 +49,9 @@ class ProducerImplBase {
virtual void flushAsync(FlushCallback callback) = 0;
virtual bool isConnected() const = 0;
virtual uint64_t getNumberOfConnectedProducer() = 0;
+
+ protected:
+ ProducerInterceptorsPtr interceptors_;
Review Comment:
Actually I do not suggest adding a field here. We'd better keep this abstract class like an interface in Java. It works because C++ supports multi inheritance.
I'd like to move this field to both `ProducerImpl` and `PartitionedProducerImpl`, then we can initialize this field in the initializer list like:
```c++
conf_(conf),
interceptors_(interceptors),
```
But I have no objection to multiple inheritance. If you keeps this design, you can add a constructor to it:
```c++
ProducerImplBase(const ProducerInterceptorsPtr& interceptors) : interceptors_(interceptors) {}
```
Then initialize it in the initializer list in `ProducerImpl` and `PartitionedProducerImpl`:
```c++
ProducerImplBase(interceptors),
conf_(conf),
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1111580752
##########
lib/ProducerInterceptors.h:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.
+ */
+
+#pragma once
+
+#include <pulsar/ProducerInterceptor.h>
+
+#include <atomic>
+#include <utility>
+#include <vector>
+
+namespace pulsar {
+
+class ProducerInterceptors {
+ public:
+ explicit ProducerInterceptors(std::vector<ProducerInterceptorPtr> interceptors)
+ : interceptors_(std::move(interceptors)) {}
Review Comment:
It seems that it would be better to apply the `clang-tidy` check into the CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1066671800
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
Review Comment:
Yes. I see the advantage.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1065688527
##########
lib/PartitionedProducerImpl.cc:
##########
@@ -443,6 +445,7 @@ void PartitionedProducerImpl::handleGetPartitions(Result result,
producers_.push_back(producer);
}
// `runPartitionUpdateTask()` will be called in `handleSinglePartitionProducerCreated()`
+ onPartitionsChange(getTopic(), newNumPartitions);
Review Comment:
Unlock `producersLock` before `onPartitionsChange` because it might took long time and it doesn't need to be protected by `producersLock`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1065682206
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
Review Comment:
An `initializer_list` as the parameter is strange. Adding an `addInterceptor` API could be better and easier to use.
```c++
ProducerConfiguration& addInterceptor(const ProducerInterceptorPtr& interceptor);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1111570587
##########
lib/ProducerImplBase.h:
##########
@@ -48,6 +49,9 @@ class ProducerImplBase {
virtual void flushAsync(FlushCallback callback) = 0;
virtual bool isConnected() const = 0;
virtual uint64_t getNumberOfConnectedProducer() = 0;
+
+ protected:
+ ProducerInterceptorsPtr interceptors_;
Review Comment:
Keeping it as an `interface` makes sense to me. I have moved them to `ProduerImpl` and `PartitonedProducerImpl`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by "RobertIndie (via GitHub)" <gi...@apache.org>.
RobertIndie commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1111611234
##########
lib/ProducerInterceptors.h:
##########
@@ -0,0 +1,56 @@
+/**
+ * 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.
+ */
+
+#pragma once
+
+#include <pulsar/ProducerInterceptor.h>
+
+#include <atomic>
+#include <utility>
+#include <vector>
+
+namespace pulsar {
+
+class ProducerInterceptors {
+ public:
+ explicit ProducerInterceptors(std::vector<ProducerInterceptorPtr> interceptors)
+ : interceptors_(std::move(interceptors)) {}
Review Comment:
> It seems that it would be better to apply the `clang-tidy` check into the CI.
+1. I have created a issue to track it: https://github.com/apache/pulsar-client-cpp/issues/200
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1066672106
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
Review Comment:
Why not use a `std::vector`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] RobertIndie commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
RobertIndie commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1066656516
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
Review Comment:
This is to make it consistent with [the java API](https://github.com/apache/pulsar/blob/932023da50a124384198778bf0b7daa8bf9a3ba4/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java#L552).
Using it, the user can easily add the interceptors by this:
```cpp
conf.intercept({interceptor1, interceptor2, interceptor3});
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar-client-cpp] BewareMyPower commented on a diff in pull request #169: [feat] Add producer interceptor
Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #169:
URL: https://github.com/apache/pulsar-client-cpp/pull/169#discussion_r1066698932
##########
include/pulsar/ProducerConfiguration.h:
##########
@@ -532,6 +533,10 @@ class PULSAR_PUBLIC ProducerConfiguration {
*/
ProducerAccessMode getAccessMode() const;
+ ProducerConfiguration& intercept(const std::initializer_list<ProducerInterceptorPtr> interceptors);
Review Comment:
I just missed the code here. Generally, `std::initializer_list` should be passed by value. BTW, the `const` could be removed.
See `std::vector`'s 10th constructor: https://en.cppreference.com/w/cpp/container/vector/vector
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org