/*-
 * #%L
 * %%
 * Copyright (C) 2018 BMW Car IT GmbH
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * #L%
 */
package io.joynr.examples.messagepersistence.consumer;

import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.joynr.arbitration.DiscoveryQos;
import io.joynr.arbitration.DiscoveryScope;
import io.joynr.exceptions.JoynrRuntimeException;
import io.joynr.messaging.MessagingQos;
import io.joynr.messaging.MessagingQosEffort;
import io.joynr.proxy.Callback;
import io.joynr.proxy.ProxyBuilder;
import io.joynr.runtime.AbstractJoynrApplication;
import joynr.vehicle.RadioAsync;
import joynr.vehicle.RadioProxy;

public class ConsumerApplication extends AbstractJoynrApplication {
    private static final Logger logger = LoggerFactory.getLogger(ConsumerApplication.class);
    private static final int NUM_REQUEST = 100;

    @Override
    public void run() {
        logger.info("ConsumerApplication running ...");

        ProxyBuilder<RadioProxy> proxyBuilder = runtime.getProxyBuilder("io.joynr.examples.messagepersistence",
                                                                        RadioProxy.class);
        DiscoveryQos discoveryQos = new DiscoveryQos();
        discoveryQos.setDiscoveryScope(DiscoveryScope.GLOBAL_ONLY);
        discoveryQos.setDiscoveryTimeoutMs(60000L);
        discoveryQos.setRetryIntervalMs(500L);
        proxyBuilder.setDiscoveryQos(discoveryQos);

        MessagingQos messagingQos = new MessagingQos();
        messagingQos.setTtl_ms(10 * 60000L); //10min in order to have time to stop/restart provider side
        messagingQos.setEffort(MessagingQosEffort.NORMAL);

        proxyBuilder.setMessagingQos(messagingQos);
        RadioAsync radioProxy = proxyBuilder.build();

        final CountDownLatch responsesCountDown = new CountDownLatch(NUM_REQUEST);

        for (int i = 0; i < NUM_REQUEST; i++) {
            radioProxy.shuffleStations(new Callback<Void>() {

                @Override
                public synchronized void onFailure(JoynrRuntimeException runtimeException) {
                    responsesCountDown.countDown();
                    logger.error("Unexpected error (" + (NUM_REQUEST - responsesCountDown.getCount()) + "/"
                            + NUM_REQUEST + ")", runtimeException);
                }

                @Override
                public synchronized void onSuccess(Void result) {
                    responsesCountDown.countDown();
                    logger.info("Shuffle stations call was successfully executed ("
                            + (NUM_REQUEST - responsesCountDown.getCount()) + "/" + NUM_REQUEST + ")");
                }
            });
        }

        logger.info("{} requests successfully sent to provider", NUM_REQUEST);

        try {
            responsesCountDown.await();
            logger.info("All {} responses arrived", NUM_REQUEST);
        } catch (InterruptedException e) {
            logger.info("Consumer application interrupted while waiting for all responses", e);
        }
    }
}

@@ -720,16 +720,54 @@
 wnLatch;
+%0Aimport java.util.concurrent.TimeUnit;
 %0A%0Aimport
@@ -1433,16 +1433,133 @@
 T = 100;
+%0A    // 10min in order to have time to stop/restart provider side%0A    private static final long TTL_MS = 10 * 60000L;
 %0A%0A    @O
@@ -2217,81 +2217,16 @@
 _ms(
-10 * 60000L); //10min in order to have time to stop/restart provider side
+TTL_MS);
 %0A   
@@ -3425,32 +3425,36 @@
 y %7B%0A            
+if (
 responsesCountDo
@@ -3462,19 +3462,54 @@
 n.await(
-);%0A
+TTL_MS, TimeUnit.MILLISECONDS)) %7B%0A    
         
@@ -3566,16 +3566,372 @@
 QUEST);%0A
+            %7D else %7B%0A                final long remainingResponses = responsesCountDown.getCount();%0A                logger.error(%22%7B%7D of %7B%7D responses arrived (%7B%7D did not arrive in time)%22,%0A                             NUM_REQUEST - remainingResponses,%0A                             NUM_REQUEST,%0A                             remainingResponses);%0A            %7D%0A
         

Use timeout in message-persistence-consumer to wait for replies