[
https://issues.apache.org/jira/browse/FLINK-8468?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16505765#comment-16505765
]
ASF GitHub Bot commented on FLINK-8468:
---------------------------------------
Github user GJL commented on a diff in the pull request:
https://github.com/apache/flink/pull/5410#discussion_r193965626
--- Diff:
flink-connectors/flink-connector-rabbitmq/src/main/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSink.java
---
@@ -105,7 +155,24 @@ public void invoke(IN value) {
try {
byte[] msg = schema.serialize(value);
- channel.basicPublish("", queueName, null, msg);
+ if (publishOptions == null) {
+ channel.basicPublish("", queueName, null, msg);
+ } else {
+ boolean mandatory =
publishOptions.computeMandatory(value);
+ boolean immediate =
publishOptions.computeImmediate(value);
+
+ if (returnListener == null && (mandatory ||
immediate)) {
+ throw new
IllegalStateException("Setting mandatory and/or immediate flags to true
requires a ReturnListener.");
+ } else {
+ String rk =
publishOptions.computeRoutingKey(value);
+ String exchange =
publishOptions.computeExchange(value);
+
+ channel.basicPublish((exchange != null)
? exchange : "",
--- End diff --
What do you think about not allowing the return of `computeRoutingKey` and
`computeExchange` to be `null`. `basicPublish` does not allow `null`s and will
throw an exception. This way our interfaces will be consistent with the ones
from RMQ.
> Make the connector to take advantage of AMQP features (routing key, exchange
> and message properties)
> ----------------------------------------------------------------------------------------------------
>
> Key: FLINK-8468
> URL: https://issues.apache.org/jira/browse/FLINK-8468
> Project: Flink
> Issue Type: Improvement
> Components: RabbitMQ Connector
> Affects Versions: 1.4.0
> Reporter: Ph.Duveau
> Priority: Major
>
> Make the connector to take advantage of AMQP features by adding a constructor
> and an interface to implement
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
