opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/utils/MessageCollectorActor.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.cluster.raft.utils;
  10
  11 import akka.actor.ActorRef;
  12 import akka.actor.Props;
  13 import akka.actor.UntypedActor;
  14 import akka.pattern.Patterns;
  15 import akka.util.Timeout;
  16 import com.google.common.base.Predicate;
  17 import com.google.common.base.Predicates;
  18 import com.google.common.base.Throwables;
  19 import com.google.common.collect.Iterables;
  20 import com.google.common.util.concurrent.Uninterruptibles;
  21 import java.util.ArrayList;
  22 import java.util.Collections;
  23 import java.util.List;
  24 import java.util.concurrent.TimeUnit;
  25 import java.util.concurrent.TimeoutException;
  26 import org.junit.Assert;
  27 import scala.concurrent.Await;
  28 import scala.concurrent.Future;
  29 import scala.concurrent.duration.Duration;
  30 import scala.concurrent.duration.FiniteDuration;
  31
  32 public class MessageCollectorActor extends UntypedActor {
  33     private static final String ARE_YOU_READY = "ARE_YOU_READY";
  34     public static final String GET_ALL_MESSAGES = "messages";
  35     private static final String CLEAR_MESSAGES = "clear-messages";
  36
  37     private final List<Object> messages = new ArrayList<>();
  38
  39     @Override public void onReceive(final Object message) throws Exception {
  40         if (ARE_YOU_READY.equals(message)) {
  41             getSender().tell("yes", getSelf());
  42         } else if (GET_ALL_MESSAGES.equals(message)) {
  43             getSender().tell(new ArrayList<>(messages), getSelf());
  44         } else if (CLEAR_MESSAGES.equals(message)) {
  45             clear();
  46         } else if (message != null) {
  47             messages.add(message);
  48         }
  49     }
  50
  51     public void clear() {
  52         messages.clear();
  53     }
  54
  55     @SuppressWarnings({"unchecked", "checkstyle:illegalCatch"})
  56     private static List<Object> getAllMessages(final ActorRef actor) {
  57         FiniteDuration operationDuration = Duration.create(5, TimeUnit.SECONDS);
  58         Timeout operationTimeout = new Timeout(operationDuration);
  59         Future<Object> future = Patterns.ask(actor, GET_ALL_MESSAGES, operationTimeout);
  60
  61         try {
  62             return (List<Object>) Await.result(future, operationDuration);
  63         } catch (RuntimeException e) {
  64             throw e;
  65         } catch (Exception e) {
  66             throw new RuntimeException(e);
  67         }
  68     }
  69
  70     public static void clearMessages(final ActorRef actor) {
  71         actor.tell(CLEAR_MESSAGES, ActorRef.noSender());
  72     }
  73
  74     /**
  75      * Get the first message that matches the specified class.
  76      *
  77      * @param actor the MessageCollectorActor reference
  78      * @param clazz the class to match
  79      * @return the first matching message
  80      */
  81     public static <T> T getFirstMatching(final ActorRef actor, final Class<T> clazz) {
  82         List<Object> allMessages = getAllMessages(actor);
  83
  84         for (Object message : allMessages) {
  85             if (message.getClass().equals(clazz)) {
  86                 return clazz.cast(message);
  87             }
  88         }
  89
  90         return null;
  91     }
  92
  93     @SuppressWarnings("checkstyle:IllegalCatch")
  94     public static <T> List<T> expectMatching(final ActorRef actor, final Class<T> clazz, final int count) {
  95         return expectMatching(actor, clazz, count, msg -> true);
  96     }
  97
  98     @SuppressWarnings("checkstyle:IllegalCatch")
  99     public static <T> List<T> expectMatching(final ActorRef actor, final Class<T> clazz, final int count,
 100             final Predicate<T> matcher) {
 101         int timeout = 5000;
 102         Exception lastEx = null;
 103         List<T> messages = Collections.emptyList();
 104         for (int i = 0; i < timeout / 50; i++) {
 105             try {
 106                 messages = getAllMatching(actor, clazz);
 107                 Iterables.removeIf(messages, Predicates.not(matcher));
 108                 if (messages.size() >= count) {
 109                     return messages;
 110                 }
 111
 112                 lastEx = null;
 113             } catch (Exception e)  {
 114                 lastEx = e;
 115             }
 116
 117             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 118         }
 119
 120         throw new AssertionError(String.format("Expected %d messages of type %s. Actual received was %d: %s", count,
 121                 clazz, messages.size(), messages), lastEx);
 122     }
 123
 124     public static <T> T expectFirstMatching(final ActorRef actor, final Class<T> clazz) {
 125         return expectFirstMatching(actor, clazz, 5000);
 126     }
 127
 128     @SuppressWarnings("checkstyle:IllegalCatch")
 129     public static <T> T expectFirstMatching(final ActorRef actor, final Class<T> clazz, final long timeout) {
 130         Exception lastEx = null;
 131         int count = (int) (timeout / 50);
 132         for (int i = 0; i < count; i++) {
 133             try {
 134                 T message = getFirstMatching(actor, clazz);
 135                 if (message != null) {
 136                     return message;
 137                 }
 138
 139                 lastEx = null;
 140             } catch (Exception e) {
 141                 lastEx = e;
 142             }
 143
 144             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 145         }
 146
 147         throw new AssertionError("Did not receive message of type " + clazz, lastEx);
 148     }
 149
 150     @SuppressWarnings("checkstyle:IllegalCatch")
 151     public static <T> T expectFirstMatching(final ActorRef actor, final Class<T> clazz, final Predicate<T> matcher) {
 152         int timeout = 5000;
 153         Exception lastEx = null;
 154         T lastMessage = null;
 155         for (int i = 0; i < timeout / 50; i++) {
 156             try {
 157                 List<T> messages = getAllMatching(actor, clazz);
 158                 for (T msg : messages) {
 159                     if (matcher.apply(msg)) {
 160                         return msg;
 161                     }
 162
 163                     lastMessage = msg;
 164                 }
 165
 166                 lastEx = null;
 167             } catch (Exception e) {
 168                 lastEx = e;
 169             }
 170
 171             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 172         }
 173
 174         throw new AssertionError(String.format("Expected specific message of type %s. Last message received was: %s",
 175                 clazz, lastMessage), lastEx);
 176     }
 177
 178     public static <T> void assertNoneMatching(final ActorRef actor, final Class<T> clazz) {
 179         assertNoneMatching(actor, clazz, 5000);
 180     }
 181
 182     @SuppressWarnings("checkstyle:IllegalCatch")
 183     public static <T> void assertNoneMatching(final ActorRef actor, final Class<T> clazz, final long timeout) {
 184         Exception lastEx = null;
 185         int count = (int) (timeout / 50);
 186         for (int i = 0; i < count; i++) {
 187             try {
 188                 T message = getFirstMatching(actor, clazz);
 189                 if (message != null) {
 190                     Assert.fail("Unexpected message received" +  message.toString());
 191                     return;
 192                 }
 193
 194                 lastEx = null;
 195             } catch (Exception e) {
 196                 lastEx = e;
 197             }
 198
 199             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 200         }
 201
 202         if (lastEx != null) {
 203             Throwables.throwIfUnchecked(lastEx);
 204             throw new RuntimeException(lastEx);
 205         }
 206
 207         return;
 208     }
 209
 210
 211     public static <T> List<T> getAllMatching(final ActorRef actor, final Class<T> clazz) {
 212         List<Object> allMessages = getAllMessages(actor);
 213
 214         List<T> output = new ArrayList<>();
 215
 216         for (Object message : allMessages) {
 217             if (message.getClass().equals(clazz)) {
 218                 output.add(clazz.cast(message));
 219             }
 220         }
 221
 222         return output;
 223     }
 224
 225     public static void waitUntilReady(final ActorRef actor) throws TimeoutException, InterruptedException {
 226         long timeout = 500;
 227         FiniteDuration duration = Duration.create(timeout, TimeUnit.MILLISECONDS);
 228         for (int i = 0; i < 10; i++) {
 229             try {
 230                 Await.ready(Patterns.ask(actor, ARE_YOU_READY, timeout), duration);
 231                 return;
 232             } catch (TimeoutException e) {
 233                 // will fall through below
 234             }
 235         }
 236
 237         throw new TimeoutException("Actor not ready in time.");
 238     }
 239
 240     public static Props props() {
 241         return Props.create(MessageCollectorActor.class);
 242     }
 243 }