Merge pull request #337 from iExecBlockchainComputing/feature/dirty-5.0.1-rc

Decode 404 response from SMS & fix subscription NPE

Author: zguesmi

gradle.properties

@@ -1,4 +1,4 @@
 version=5.1.0-SNAPSHOT
-iexecCommonVersion=5.0.0
+iexecCommonVersion=5.1.0-SNAPSHOT
 nexusUser=fake
 nexusPassword=fake

src/main/java/com/iexec/worker/feign/client/SmsClient.java

@@ -28,7 +28,7 @@
 
 import feign.FeignException;
 
-@FeignClient(name = "SmsClient", url = "#{publicConfigurationService.smsURL}")
+@FeignClient(name = "SmsClient", url = "#{publicConfigurationService.smsURL}", decode404 = true)
 public interface SmsClient {
 
     @PostMapping("/untee/secrets")

src/main/java/com/iexec/worker/pubsub/StompClient.java

@@ -18,6 +18,7 @@
 
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
@@ -90,8 +91,12 @@ public StompClient(ApplicationEventPublisher applicationEventPublisher,
      * @param messageHandler an implementation of 
      * @return
      */
-    Subscription subscribeToTopic(String topic, StompFrameHandler messageHandler) {
-        return this.session.subscribe(topic, messageHandler);
+    Optional<Subscription> subscribeToTopic(String topic, StompFrameHandler messageHandler) {
+        // Should not let other threads subscribe
+        // when the session is not ready yet.
+        return this.session != null
+                ? Optional.of(this.session.subscribe(topic, messageHandler))
+                : Optional.empty();
     }
 
     @PostConstruct

src/main/java/com/iexec/worker/pubsub/SubscriptionService.java

@@ -57,17 +57,18 @@ public SubscriptionService(WorkerConfigurationService workerConfigurationService
      * 
      * @param chainTaskId id of the task to which to subscribe
      */
-    public synchronized void subscribeToTopic(String chainTaskId) {
+    public void subscribeToTopic(String chainTaskId) {
         String topic = getTaskTopicName(chainTaskId);
         if (this.chainTaskIdToSubscription.containsKey(chainTaskId)) {
             log.info("Already subscribed to topic [chainTaskId:{}, topic:{}]",
                     chainTaskId, topic);
             return;
         }
         MessageHandler messageHandler = new MessageHandler(chainTaskId, this.workerWalletAddress);
-        Subscription subscription = stompClient.subscribeToTopic(topic, messageHandler);
-        this.chainTaskIdToSubscription.put(chainTaskId, subscription);
-        log.info("Subscribed to topic [chainTaskId:{}, topic:{}]", chainTaskId, topic);
+        stompClient.subscribeToTopic(topic, messageHandler).ifPresent(subscription -> {
+            this.chainTaskIdToSubscription.put(chainTaskId, subscription);
+            log.info("Subscribed to topic [chainTaskId:{}, topic:{}]", chainTaskId, topic);
+        });
     }
 
     /**

src/test/java/com/iexec/worker/pubsub/SubscriptionServiceTests.java

@@ -36,6 +36,8 @@
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.messaging.simp.stomp.StompSession.Subscription;
 
+import java.util.Optional;
+
 public class SubscriptionServiceTests {
 
     @Mock
@@ -50,7 +52,8 @@ public class SubscriptionServiceTests {
 
     private static final String WORKER_WALLET_ADDRESS = "0x1234";
     private static final String CHAIN_TASK_ID = "chaintaskid";
-    private static final Subscription SUBSCRIPTION = mock(Subscription.class);
+    private static final Optional<Subscription> SUBSCRIPTION =
+            Optional.of(mock(Subscription.class));
 
     @Before
     public void init() {
@@ -87,13 +90,13 @@ public void shouldUnsubscribeFromTopic() {
         assertThat(subscriptionService.isSubscribedToTopic(CHAIN_TASK_ID)).isTrue();
         subscriptionService.unsubscribeFromTopic(CHAIN_TASK_ID);
         assertThat(subscriptionService.isSubscribedToTopic(CHAIN_TASK_ID)).isFalse();
-        verify(SUBSCRIPTION, times(1)).unsubscribe();
+        verify(SUBSCRIPTION.get(), times(1)).unsubscribe();
     }
 
     @Test
     public void shouldNotUnsubscribeFromInexistentTopic() {
         assertThat(subscriptionService.isSubscribedToTopic(CHAIN_TASK_ID)).isFalse();
         subscriptionService.unsubscribeFromTopic(CHAIN_TASK_ID);
-        verify(SUBSCRIPTION, never()).unsubscribe();
+        verify(SUBSCRIPTION.get(), never()).unsubscribe();
     }
 }