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