+ public void testSuccess() throws Exception {
+ BidirectionalTopicParams opParams =
+ BidirectionalTopicParams.builder().sinkTopic(MY_SINK).sourceTopic(MY_SOURCE).build();
+ config = new BidirectionalTopicConfig(blockingExecutor, opParams, topicMgr, SdnrOperation.SELECTOR_KEYS);
+
+ params = params.toBuilder().retry(0).timeoutSec(5).executor(blockingExecutor).build();
+
+ operation = new SdnrOperation(params, config);
+ operation.setProperty(OperationProperties.EVENT_PAYLOAD, "my payload");
+
+ outcome = operation.start().get();
+ assertEquals(OperationResult.SUCCESS, outcome.getResult());
+ assertTrue(outcome.getResponse() instanceof PciMessage);