fix AmqpPublisherActor

Signed-off-by: Dominik Guggemos <dominik.guggemos@bosch-si.com>
eclipse-ditto · Mar 1, 2018 · 34da389 · 34da389
1 parent d898d61
commit 34da389
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 22 deletions.
diff --git a/...rc/main/java/org/eclipse/ditto/services/amqpbridge/messaging/amqp/AmqpPublisherActor.java b/...rc/main/java/org/eclipse/ditto/services/amqpbridge/messaging/amqp/AmqpPublisherActor.java
@@ -94,10 +94,10 @@ public Receive createReceive() {
                         log.debug("Response dropped, missing replyTo address.");
                     }
                 })
-                .match(ExternalMessage.class, ExternalMessage::isCommandResponse, event -> {
+                .match(ExternalMessage.class, ExternalMessage::isEvent, event -> {
                     final String correlationId = event.getHeaders().get(CORRELATION_ID.getKey());
                     LogUtil.enhanceLogWithCorrelationId(log, correlationId);
-                    log.debug("Received command response {} ", event);
+                    log.debug("Received thing event {} ", event);
                     amqpConnection.getEventTarget().ifPresent(target -> sendMessage(target, event));
                 })
                 .matchAny(m -> {
@@ -136,7 +136,7 @@ private Message toJmsMessage(final ExternalMessage externalMessage) throws JMSEx
     @Nullable
     private MessageProducer getProducer(final String target) {
         return Optional.of(target)
-                .filter(String::isEmpty)
+                .filter(s -> !s.isEmpty())
                 .map(t -> producerMap.computeIfAbsent(target, this::createMessageProducer)).orElse(null);
     }
 
@@ -146,7 +146,7 @@ private MessageProducer createMessageProducer(final String target) {
         log.debug("Creating AMQP Producer for '{}'", target);
         try {
             return session.createProducer(destination);
-        } catch (JMSException e) {
+        } catch (final JMSException e) {
             log.warning("Could not create producer for {}.", target);
             return null;
         }

diff --git a/...c/test/java/org/eclipse/ditto/services/amqpbridge/messaging/amqp/AmqpClientActorTest.java b/...c/test/java/org/eclipse/ditto/services/amqpbridge/messaging/amqp/AmqpClientActorTest.java
@@ -19,6 +19,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.timeout;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -28,6 +29,7 @@
 import java.util.concurrent.TimeUnit;
 
 import javax.jms.Connection;
+import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
@@ -53,7 +55,6 @@
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
-import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
@@ -94,6 +95,8 @@ public class AmqpClientActorTest {
     private MessageConsumer mockConsumer = Mockito.mock(MessageConsumer.class);
     @Mock
     private MessageProducer mockProducer = Mockito.mock(MessageProducer.class);
+    @Mock
+    private TextMessage mockTextMessage = Mockito.mock(TextMessage.class);
 
     @BeforeClass
     public static void setUp() {
@@ -110,7 +113,8 @@ public static void tearDown() {
     public void init() throws JMSException {
         when(mockConnection.createSession(false, Session.AUTO_ACKNOWLEDGE)).thenReturn(mockSession);
         when(mockSession.createConsumer(any(JmsQueue.class))).thenReturn(mockConsumer);
-        when(mockSession.createProducer(any(JmsQueue.class))).thenReturn(mockProducer);
+        when(mockSession.createProducer(any(Destination.class))).thenReturn(mockProducer);
+        when(mockSession.createTextMessage(anyString())).thenReturn(mockTextMessage);
     }
 
     @Test
@@ -264,7 +268,7 @@ public void testCloseConnectionFails() throws JMSException {
     }
 
     @Test
-    public void testInitAfterTimeout() {
+    public void testInitializeClientActorAfterTimeout() {
         new TestKit(actorSystem) {{
             final String pubSubTargetPath = getRef().path().toStringWithoutAddress();
             final Props props =
@@ -286,7 +290,6 @@ public void testInitAfterTimeout() {
     }
 
     @Test
-    @Ignore("currently failing")
     public void testConsumeMessageAndExpectForwardToProxyActor() throws JMSException {
         new TestKit(actorSystem) {{
             final ActorRef mediator = DistributedPubSub.get(actorSystem).mediator();
@@ -313,12 +316,8 @@ public void testConsumeMessageAndExpectForwardToProxyActor() throws JMSException
     }
 
     @Test
-    @Ignore("currently failing")
-    public void testReceiveThingEventAndExpectForwardToAmqpClient() throws JMSException {
+    public void testReceiveThingEventAndExpectForwardToJMSProducer() throws JMSException {
         new TestKit(actorSystem) {{
-            final ActorRef mediator = DistributedPubSub.get(actorSystem).mediator();
-            mediator.tell(new DistributedPubSubMediator.Put(getRef()), getRef());
-
             final String pubSubTargetPath = getRef().path().toStringWithoutAddress();
             final Props props =
                     AmqpClientActor.props(connectionId, getRef(), amqpConnection, pubSubTargetPath,
@@ -328,17 +327,10 @@ public void testReceiveThingEventAndExpectForwardToAmqpClient() throws JMSExcept
             amqpClientActor.tell(CreateConnection.of(amqpConnection, DittoHeaders.empty()), getRef());
             expectMsg(CONNECTED_SUCCESS);
 
-            amqpClientActor.tell(TestConstants.thingModified(singletonList("")), getRef());
-
-            //TODO find better way, this is ugly
-            try {
-                Thread.sleep(1000);
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
 
-            verify(mockProducer).send(any(Message.class));
+            amqpClientActor.tell(TestConstants.thingModified(singletonList("")), getRef());
 
+            verify(mockProducer, timeout(500)).send(mockTextMessage);
         }};
     }