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     private static final String GET_ALL_MESSAGES = "get-all-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             messages.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     public 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 (T) 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     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> List<T> getAllMatching(ActorRef actor, Class<T> clazz) throws Exception {
 151         List<Object> allMessages = getAllMessages(actor);
 152
 153         List<T> output = Lists.newArrayList();
 154
 155         for(Object message : allMessages){
 156             if(message.getClass().equals(clazz)){
 157                 output.add((T) message);
 158             }
 159         }
 160
 161         return output;
 162     }
 163
 164     public static void waitUntilReady(ActorRef actor) throws Exception {
 165         long timeout = 500;
 166         FiniteDuration duration = Duration.create(timeout, TimeUnit.MILLISECONDS);
 167         for(int i = 0; i < 10; i++) {
 168             try {
 169                 Await.ready(Patterns.ask(actor, ARE_YOU_READY, timeout), duration);
 170                 return;
 171             } catch (TimeoutException e) {
 172             }
 173         }
 174
 175         throw new TimeoutException("Actor not ready in time.");
 176     }
 177
 178     public static Props props() {
 179         return Props.create(MessageCollectorActor.class);
 180     }
 181 }