RMQSink extends RichSinkFunction<IN> { @Override public void invoke(IN value) { try { byte[] msg = schema.serialize(value); channel.basicPublish("", queueName, null, msg); } catch (IOException e) { if (logFailuresOnly) { LOG.error("Cannot send RMQ message {} at {}", queueName, rmqConnectionConfig.getHost(), e); } else { throw new RuntimeException("Cannot send RMQ message " + queueName + " at " + rmqConnectionConfig.getHost(), e); } } } RMQSink(RMQConnectionConfig rmqConnectionConfig, String queueName, SerializationSchema<IN> schema); void setLogFailuresOnly(boolean logFailuresOnly); @Override void open(Configuration config); @Override void invoke(IN value); @Override void close();  }
@Test public void invokePublishBytesToQueue() throws Exception { RMQSink<String> rmqSink = createRMQSink(); rmqSink.invoke(MESSAGE_STR, SinkContextUtil.forTimestamp(0)); verify(serializationSchema).serialize(MESSAGE_STR); verify(channel).basicPublish("", QUEUE_NAME, null, MESSAGE); }
@Test(expected = RuntimeException.class) public void exceptionDuringPublishingIsNotIgnored() throws Exception { RMQSink<String> rmqSink = createRMQSink(); doThrow(IOException.class).when(channel).basicPublish("", QUEUE_NAME, null, MESSAGE); rmqSink.invoke("msg", SinkContextUtil.forTimestamp(0)); }