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 org.opendaylight.controller.cluster.raft.SerializationUtils;
  26 import scala.concurrent.Await;
  27 import scala.concurrent.Future;
  28 import scala.concurrent.duration.Duration;
  29 import scala.concurrent.duration.FiniteDuration;
  30
  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(Object message) throws Exception {
  40         if(message.equals(ARE_YOU_READY)) {
  41             getSender().tell("yes", getSelf());
  42             return;
  43         }
  44
  45         if(GET_ALL_MESSAGES.equals(message)) {
  46             getSender().tell(new ArrayList<>(messages), getSelf());
  47         } else if(CLEAR_MESSAGES.equals(message)) {
  48             clear();
  49         } else if(message != null) {
  50             messages.add(SerializationUtils.fromSerializable(message));
  51         }
  52     }
  53
  54     public void clear() {
  55         messages.clear();
  56     }
  57
  58     private static List<Object> getAllMessages(ActorRef actor) throws Exception {
  59         FiniteDuration operationDuration = Duration.create(5, TimeUnit.SECONDS);
  60         Timeout operationTimeout = new Timeout(operationDuration);
  61         Future<Object> future = Patterns.ask(actor, GET_ALL_MESSAGES, operationTimeout);
  62
  63         return (List<Object>) Await.result(future, operationDuration);
  64     }
  65
  66     public static void clearMessages(ActorRef actor) {
  67         actor.tell(CLEAR_MESSAGES, ActorRef.noSender());
  68     }
  69
  70     /**
  71      * Get the first message that matches the specified class
  72      * @param actor
  73      * @param clazz
  74      * @return
  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     public static <T> List<T> expectMatching(ActorRef actor, Class<T> clazz, int count) {
  89         int timeout = 5000;
  90         List<T> messages = Collections.emptyList();
  91         for(int i = 0; i < timeout / 50; i++) {
  92             try {
  93                 messages = getAllMatching(actor, clazz);
  94                 if(messages.size() >= count) {
  95                     return messages;
  96                 }
  97             } catch (Exception e) {}
  98
  99             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 100         }
 101
 102         Assert.fail(String.format("Expected %d messages of type %s. Actual received was %d: %s", count, clazz,
 103                 messages.size(), messages));
 104         return null;
 105     }
 106
 107     public static <T> T expectFirstMatching(ActorRef actor, Class<T> clazz) {
 108         return expectFirstMatching(actor, clazz, 5000);
 109     }
 110
 111
 112     public static <T> T expectFirstMatching(ActorRef actor, Class<T> clazz, long timeout) {
 113         int count = (int) (timeout / 50);
 114         for(int i = 0; i < count; i++) {
 115             try {
 116                 T message = getFirstMatching(actor, clazz);
 117                 if(message != null) {
 118                     return message;
 119                 }
 120             } catch (Exception e) {}
 121
 122             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 123         }
 124
 125         Assert.fail("Did not receive message of type " + clazz);
 126         return null;
 127     }
 128
 129     public static <T> T expectFirstMatching(ActorRef actor, Class<T> clazz, Predicate<T> matcher) {
 130         int timeout = 5000;
 131         T lastMessage = null;
 132         for(int i = 0; i < timeout / 50; i++) {
 133             try {
 134                 List<T> messages = getAllMatching(actor, clazz);
 135                 for(T msg: messages) {
 136                     if(matcher.apply(msg)) {
 137                         return msg;
 138                     }
 139
 140                     lastMessage = msg;
 141                 }
 142             } catch (Exception e) {}
 143
 144             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 145         }
 146
 147         Assert.fail(String.format("Expected specific message of type %s. Last message received was: %s", clazz, lastMessage));
 148         return null;
 149     }
 150
 151     public static <T> void assertNoneMatching(ActorRef actor, Class<T> clazz) {
 152         assertNoneMatching(actor, clazz, 5000);
 153     }
 154
 155     public static <T> void assertNoneMatching(ActorRef actor, Class<T> clazz, long timeout) {
 156         int count = (int) (timeout / 50);
 157         for(int i = 0; i < count; i++) {
 158             try {
 159                 T message = getFirstMatching(actor, clazz);
 160                 if(message != null) {
 161                     Assert.fail("Unexpected message received" +  message.toString());
 162                     return;
 163                 }
 164             } catch (Exception e) {}
 165
 166             Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS);
 167         }
 168
 169         return;
 170     }
 171
 172
 173     public static <T> List<T> getAllMatching(ActorRef actor, Class<T> clazz) throws Exception {
 174         List<Object> allMessages = getAllMessages(actor);
 175
 176         List<T> output = Lists.newArrayList();
 177
 178         for(Object message : allMessages){
 179             if(message.getClass().equals(clazz)){
 180                 output.add(clazz.cast(message));
 181             }
 182         }
 183
 184         return output;
 185     }
 186
 187     public static void waitUntilReady(ActorRef actor) throws Exception {
 188         long timeout = 500;
 189         FiniteDuration duration = Duration.create(timeout, TimeUnit.MILLISECONDS);
 190         for(int i = 0; i < 10; i++) {
 191             try {
 192                 Await.ready(Patterns.ask(actor, ARE_YOU_READY, timeout), duration);
 193                 return;
 194             } catch (TimeoutException e) {
 195             }
 196         }
 197
 198         throw new TimeoutException("Actor not ready in time.");
 199     }
 200
 201     public static Props props() {
 202         return Props.create(MessageCollectorActor.class);
 203     }
 204 }