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.collect.Lists;
  18 import com.google.common.util.concurrent.Uninterruptibles;
  19 import java.util.ArrayList;
  20 import java.util.Collections;
  21 import java.util.List;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.TimeoutException;
  24 import org.junit.Assert;
  25 import scala.concurrent.Await;
  26 import scala.concurrent.Future;
  27 import scala.concurrent.duration.Duration;
  28 import scala.concurrent.duration.FiniteDuration;
  29
  30
  31 public class MessageCollectorActor extends UntypedActor {
  32     private static final String ARE_YOU_READY = "ARE_YOU_READY";
  33     public static final String GET_ALL_MESSAGES = "messages";
  34     private static final String CLEAR_MESSAGES = "clear-messages";
  35
  36     private final List<Object> messages = new ArrayList<>();
  37
  38     @Override public void onReceive(Object message) throws Exception {
  39         if(message.equals(ARE_YOU_READY)) {
  40             getSender().tell("yes", getSelf());
  41             return;
  42         }
  43
  44         if(GET_ALL_MESSAGES.equals(message)) {
  45             getSender().tell(new ArrayList<>(messages), getSelf());
  46         } else if(CLEAR_MESSAGES.equals(message)) {
  47             clear();
  48         } else if(message != null) {
  49             messages.add(message);
  50         }
  51     }
  52
  53     public void clear() {
  54         messages.clear();
  55     }
  56
  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      * @param actor
  72      * @param clazz
  73      * @return
  74      */
  75     public static <T> T getFirstMatching(ActorRef actor, Class<T> clazz) throws Exception {
  76         List<Object> allMessages = getAllMessages(actor);
  77
  78         for(Object message : allMessages){
  79             if(message.getClass().equals(clazz)){
  80                 return clazz.cast(message);
  81             }
  82         }
  83
  84         return null;
  85     }
  86
  87     public static <T> List<T> expectMatching(ActorRef actor, Class<T> clazz, int count) {
  88         int timeout = 5000;
  89         List<T> messages = Collections.emptyList();
  90         for(int i = 0; i < timeout / 50; i++) {
  91             try {
  92                 messages = getAllMatching(actor, clazz);
  93                 if(messages.size() >= count) {
  94                     return messages;
  95                 }
  96             } catch (Exception e) {}
  97
  98             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
  99         }
 100
 101         Assert.fail(String.format("Expected %d messages of type %s. Actual received was %d: %s", count, clazz,
 102                 messages.size(), messages));
 103         return null;
 104     }
 105
 106     public static <T> T expectFirstMatching(ActorRef actor, Class<T> clazz) {
 107         return expectFirstMatching(actor, clazz, 5000);
 108     }
 109
 110
 111     public static <T> T expectFirstMatching(ActorRef actor, Class<T> clazz, long timeout) {
 112         int count = (int) (timeout / 50);
 113         for(int i = 0; i < count; i++) {
 114             try {
 115                 T message = getFirstMatching(actor, clazz);
 116                 if(message != null) {
 117                     return message;
 118                 }
 119             } catch (Exception e) {}
 120
 121             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 122         }
 123
 124         Assert.fail("Did not receive message of type " + clazz);
 125         return null;
 126     }
 127
 128     public static <T> T expectFirstMatching(ActorRef actor, Class<T> clazz, Predicate<T> matcher) {
 129         int timeout = 5000;
 130         T lastMessage = null;
 131         for(int i = 0; i < timeout / 50; i++) {
 132             try {
 133                 List<T> messages = getAllMatching(actor, clazz);
 134                 for(T msg: messages) {
 135                     if(matcher.apply(msg)) {
 136                         return msg;
 137                     }
 138
 139                     lastMessage = msg;
 140                 }
 141             } catch (Exception e) {}
 142
 143             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 144         }
 145
 146         Assert.fail(String.format("Expected specific message of type %s. Last message received was: %s", clazz, lastMessage));
 147         return null;
 148     }
 149
 150     public static <T> void assertNoneMatching(ActorRef actor, Class<T> clazz) {
 151         assertNoneMatching(actor, clazz, 5000);
 152     }
 153
 154     public static <T> void assertNoneMatching(ActorRef actor, Class<T> clazz, long timeout) {
 155         int count = (int) (timeout / 50);
 156         for(int i = 0; i < count; i++) {
 157             try {
 158                 T message = getFirstMatching(actor, clazz);
 159                 if(message != null) {
 160                     Assert.fail("Unexpected message received" +  message.toString());
 161                     return;
 162                 }
 163             } catch (Exception e) {}
 164
 165             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 166         }
 167
 168         return;
 169     }
 170
 171
 172     public static <T> List<T> getAllMatching(ActorRef actor, Class<T> clazz) throws Exception {
 173         List<Object> allMessages = getAllMessages(actor);
 174
 175         List<T> output = Lists.newArrayList();
 176
 177         for(Object message : allMessages){
 178             if(message.getClass().equals(clazz)){
 179                 output.add(clazz.cast(message));
 180             }
 181         }
 182
 183         return output;
 184     }
 185
 186     public static void waitUntilReady(ActorRef actor) throws Exception {
 187         long timeout = 500;
 188         FiniteDuration duration = Duration.create(timeout, TimeUnit.MILLISECONDS);
 189         for(int i = 0; i < 10; i++) {
 190             try {
 191                 Await.ready(Patterns.ask(actor, ARE_YOU_READY, timeout), duration);
 192                 return;
 193             } catch (TimeoutException e) {
 194             }
 195         }
 196
 197         throw new TimeoutException("Actor not ready in time.");
 198     }
 199
 200     public static Props props() {
 201         return Props.create(MessageCollectorActor.class);
 202     }
 203 }