opendaylight/md-sal/sal-distributed-datastore/src/test/java/org/opendaylight/controller/cluster/datastore/actors/client/SequencedQueueTest.java

   1 /*
   2  * Copyright (c) 2016 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 package org.opendaylight.controller.cluster.datastore.actors.client;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertFalse;
  12 import static org.junit.Assert.assertNotNull;
  13 import static org.junit.Assert.assertNull;
  14 import static org.junit.Assert.assertSame;
  15 import static org.junit.Assert.assertTrue;
  16 import static org.mockito.Matchers.any;
  17 import static org.mockito.Mockito.doReturn;
  18 import static org.mockito.Mockito.verify;
  19 import static org.mockito.Mockito.verifyNoMoreInteractions;
  20 import akka.actor.ActorRef;
  21 import akka.actor.ActorSystem;
  22 import akka.testkit.TestProbe;
  23 import java.util.Optional;
  24 import java.util.concurrent.CompletableFuture;
  25 import java.util.concurrent.ThreadLocalRandom;
  26 import java.util.concurrent.TimeUnit;
  27 import org.junit.After;
  28 import org.junit.AfterClass;
  29 import org.junit.Before;
  30 import org.junit.BeforeClass;
  31 import org.junit.Test;
  32 import org.mockito.ArgumentCaptor;
  33 import org.mockito.Mock;
  34 import org.mockito.MockitoAnnotations;
  35 import org.opendaylight.controller.cluster.access.ABIVersion;
  36 import org.opendaylight.controller.cluster.access.concepts.AbstractRequestFailureProxy;
  37 import org.opendaylight.controller.cluster.access.concepts.AbstractRequestProxy;
  38 import org.opendaylight.controller.cluster.access.concepts.Request;
  39 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  40 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
  41 import org.opendaylight.controller.cluster.access.concepts.Response;
  42 import org.opendaylight.controller.cluster.common.actor.TestTicker;
  43 import org.opendaylight.yangtools.concepts.WritableIdentifier;
  44 import scala.concurrent.duration.FiniteDuration;
  45
  46 /**
  47  * Test suite covering logic contained in {@link SequencedQueue}. It assumes {@link SequencedQueueEntryTest} passes.
  48  *
  49  * @author Robert Varga
  50  */
  51 public class SequencedQueueTest {
  52     private static class MockFailure extends RequestFailure<WritableIdentifier, MockFailure> {
  53         private static final long serialVersionUID = 1L;
  54
  55         MockFailure(final WritableIdentifier target, final long sequence, final long retry, final RequestException cause) {
  56             super(target, sequence, retry, cause);
  57         }
  58
  59         @Override
  60         protected AbstractRequestFailureProxy<WritableIdentifier, MockFailure> externalizableProxy(final ABIVersion version) {
  61             return null;
  62         }
  63
  64         @Override
  65         protected MockFailure cloneAsVersion(final ABIVersion version) {
  66             return this;
  67         }
  68     }
  69
  70     private static class MockRequest extends Request<WritableIdentifier, MockRequest> {
  71         private static final long serialVersionUID = 1L;
  72
  73         MockRequest(final WritableIdentifier target, final long sequence, final ActorRef replyTo) {
  74             super(target, sequence, 0, replyTo);
  75         }
  76
  77
  78         MockRequest(final MockRequest request, final long retry) {
  79             super(request, retry);
  80         }
  81
  82         @Override
  83         public RequestFailure<WritableIdentifier, ?> toRequestFailure(final RequestException cause) {
  84             return new MockFailure(getTarget(), getSequence(), getRetry(), cause);
  85         }
  86
  87         @Override
  88         protected AbstractRequestProxy<WritableIdentifier, MockRequest> externalizableProxy(final ABIVersion version) {
  89             return null;
  90         }
  91
  92         @Override
  93         protected MockRequest cloneAsVersion(final ABIVersion version) {
  94             return this;
  95         }
  96
  97         @Override
  98         protected MockRequest cloneAsRetry(final long retry) {
  99             return new MockRequest(this, retry);
 100         }
 101     };
 102
 103     @Mock
 104     private ActorRef mockReplyTo;
 105     @Mock
 106     private WritableIdentifier mockIdentifier;
 107     @Mock
 108     private RequestException mockCause;
 109     @Mock
 110     private RequestCallback mockCallback;
 111     @Mock
 112     private ClientActorBehavior mockBehavior;
 113
 114     private TestTicker ticker;
 115     private BackendInfo mockBackendInfo;
 116     private MockRequest mockRequest;
 117     private MockRequest mockRequest2;
 118     private Response<WritableIdentifier, ?> mockResponse;
 119     private Long mockCookie;
 120
 121     private static ActorSystem actorSystem;
 122     private TestProbe mockActor;
 123
 124     private SequencedQueue queue;
 125
 126     @BeforeClass
 127     public static void setupClass() {
 128         actorSystem = ActorSystem.apply();
 129     }
 130
 131     @AfterClass
 132     public static void teardownClass() {
 133         actorSystem.terminate();
 134     }
 135
 136     @Before
 137     public void setup() {
 138         MockitoAnnotations.initMocks(this);
 139
 140         doReturn(mockBehavior).when(mockCallback).complete(any(MockFailure.class));
 141
 142         ticker = new TestTicker();
 143         ticker.increment(ThreadLocalRandom.current().nextLong());
 144
 145         mockActor = TestProbe.apply(actorSystem);
 146         mockBackendInfo = new BackendInfo(mockActor.ref(), ABIVersion.current());
 147         mockRequest = new MockRequest(mockIdentifier, ThreadLocalRandom.current().nextLong(), mockReplyTo);
 148         mockRequest2 = new MockRequest(mockIdentifier, mockRequest.getSequence() + 1, mockReplyTo);
 149         mockResponse = mockRequest.toRequestFailure(mockCause);
 150         mockCookie = ThreadLocalRandom.current().nextLong();
 151
 152         queue = new SequencedQueue(mockCookie, ticker);
 153     }
 154
 155     @After
 156     public void teardown() {
 157         actorSystem.stop(mockActor.ref());
 158     }
 159
 160     @Test
 161     public void testGetCookie() {
 162         assertSame(mockCookie, queue.getCookie());
 163     }
 164
 165     @Test
 166     public void testEmptyClose() {
 167         assertFalse(queue.hasCompleted());
 168         queue.close();
 169         assertTrue(queue.hasCompleted());
 170     }
 171
 172     @Test(expected=IllegalStateException.class)
 173     public void testClosedEnqueueRequest() {
 174         queue.close();
 175
 176         // Kaboom
 177         queue.enqueueRequest(mockRequest, mockCallback);
 178     }
 179
 180     @Test
 181     public void testCloseIdempotent() {
 182         queue.close();
 183         queue.close();
 184     }
 185
 186     @Test
 187     public void testPoison() {
 188         queue.enqueueRequest(mockRequest, mockCallback);
 189         queue.poison(mockCause);
 190
 191         final ArgumentCaptor<MockFailure> captor = ArgumentCaptor.forClass(MockFailure.class);
 192         verify(mockCallback).complete(captor.capture());
 193         assertSame(mockCause, captor.getValue().getCause());
 194     }
 195
 196     @Test(expected=IllegalStateException.class)
 197     public void testPoisonPerformsClose() {
 198         // Implies close()
 199         queue.poison(mockCause);
 200
 201         // Kaboom
 202         queue.enqueueRequest(mockRequest, mockCallback);
 203     }
 204
 205     @Test
 206     public void testPoisonIdempotent() {
 207         queue.poison(mockCause);
 208         queue.poison(mockCause);
 209     }
 210
 211     @Test
 212     public void testEnqueueRequestNeedsBackend() {
 213         final Optional<FiniteDuration> ret = queue.enqueueRequest(mockRequest, mockCallback);
 214
 215         assertNotNull(ret);
 216         assertFalse(ret.isPresent());
 217     }
 218
 219     @Test
 220     public void testExpectProof() {
 221         final CompletableFuture<BackendInfo> proof = new CompletableFuture<>();
 222         assertTrue(queue.expectProof(proof));
 223         assertFalse(queue.expectProof(proof));
 224     }
 225
 226     @Test(expected=NullPointerException.class)
 227     public void testSetBackendNull() {
 228         final CompletableFuture<BackendInfo> proof = new CompletableFuture<>();
 229         assertTrue(queue.expectProof(proof));
 230         queue.setBackendInfo(proof, null);
 231     }
 232
 233     @Test
 234     public void testSetBackendWithNoResolution() {
 235         queue.enqueueRequest(mockRequest, mockCallback);
 236
 237         final CompletableFuture<BackendInfo> proof = new CompletableFuture<>();
 238         final Optional<FiniteDuration> ret = queue.setBackendInfo(proof, mockBackendInfo);
 239         assertNotNull(ret);
 240         assertFalse(ret.isPresent());
 241     }
 242
 243     @Test
 244     public void testSetBackendWithWrongProof() {
 245         queue.enqueueRequest(mockRequest, mockCallback);
 246
 247         final CompletableFuture<BackendInfo> proof = new CompletableFuture<>();
 248         assertTrue(queue.expectProof(proof));
 249
 250         final Optional<FiniteDuration> ret = queue.setBackendInfo(new CompletableFuture<>(), mockBackendInfo);
 251         assertNotNull(ret);
 252         assertFalse(ret.isPresent());
 253     }
 254
 255     @Test
 256     public void testSetBackendWithNoRequests() {
 257         // this utility method covers the entire test
 258         setupBackend();
 259     }
 260
 261     @Test
 262     public void testSetbackedWithRequestsNoTimer() {
 263         queue.enqueueRequest(mockRequest, mockCallback);
 264
 265         final CompletableFuture<BackendInfo> proof = new CompletableFuture<>();
 266         assertTrue(queue.expectProof(proof));
 267         assertFalse(mockActor.msgAvailable());
 268
 269         final Optional<FiniteDuration> ret = queue.setBackendInfo(proof, mockBackendInfo);
 270         assertNotNull(ret);
 271         assertTrue(ret.isPresent());
 272
 273         assertTransmit(mockRequest);
 274     }
 275
 276     @Test
 277     public void testEnqueueRequestNeedsTimer() {
 278         setupBackend();
 279
 280         final Optional<FiniteDuration> ret = queue.enqueueRequest(mockRequest, mockCallback);
 281         assertNotNull(ret);
 282         assertTrue(ret.isPresent());
 283         assertTransmit(mockRequest);
 284     }
 285
 286     @Test
 287     public void testEnqueueRequestWithoutTimer() {
 288         setupBackend();
 289
 290         // First request
 291         Optional<FiniteDuration> ret = queue.enqueueRequest(mockRequest, mockCallback);
 292         assertNotNull(ret);
 293         assertTrue(ret.isPresent());
 294         assertTransmit(mockRequest);
 295
 296         // Second request, no timer fired
 297         ret = queue.enqueueRequest(mockRequest2, mockCallback);
 298         assertNull(ret);
 299         assertTransmit(mockRequest2);
 300     }
 301
 302     @Test
 303     public void testRunTimeoutEmpty() throws NoProgressException {
 304         final boolean ret = queue.runTimeout();
 305         assertFalse(ret);
 306     }
 307
 308     @Test
 309     public void testRunTimeoutWithoutShift() throws NoProgressException {
 310         queue.enqueueRequest(mockRequest, mockCallback);
 311         final boolean ret = queue.runTimeout();
 312         assertFalse(ret);
 313     }
 314
 315     @Test
 316     public void testRunTimeoutWithTimeoutLess() throws NoProgressException {
 317         queue.enqueueRequest(mockRequest, mockCallback);
 318
 319         ticker.increment(SequencedQueue.REQUEST_TIMEOUT_NANOS - 1);
 320
 321         final boolean ret = queue.runTimeout();
 322         assertFalse(ret);
 323     }
 324
 325     @Test
 326     public void testRunTimeoutWithTimeoutExact() throws NoProgressException {
 327         queue.enqueueRequest(mockRequest, mockCallback);
 328
 329         ticker.increment(SequencedQueue.REQUEST_TIMEOUT_NANOS);
 330
 331         final boolean ret = queue.runTimeout();
 332         assertTrue(ret);
 333     }
 334
 335     @Test
 336     public void testRunTimeoutWithTimeoutMore() throws NoProgressException {
 337         queue.enqueueRequest(mockRequest, mockCallback);
 338
 339         ticker.increment(SequencedQueue.REQUEST_TIMEOUT_NANOS + 1);
 340
 341         final boolean ret = queue.runTimeout();
 342         assertTrue(ret);
 343     }
 344
 345     @Test(expected=NoProgressException.class)
 346     public void testRunTimeoutWithoutProgressExact() throws NoProgressException {
 347         queue.enqueueRequest(mockRequest, mockCallback);
 348
 349         ticker.increment(SequencedQueue.NO_PROGRESS_TIMEOUT_NANOS);
 350
 351         // Kaboom
 352         queue.runTimeout();
 353     }
 354
 355     @Test(expected=NoProgressException.class)
 356     public void testRunTimeoutWithoutProgressMore() throws NoProgressException {
 357         queue.enqueueRequest(mockRequest, mockCallback);
 358
 359         ticker.increment(SequencedQueue.NO_PROGRESS_TIMEOUT_NANOS + 1);
 360
 361         // Kaboom
 362         queue.runTimeout();
 363     }
 364
 365     @Test
 366     public void testRunTimeoutEmptyWithoutProgressExact() throws NoProgressException {
 367         ticker.increment(SequencedQueue.NO_PROGRESS_TIMEOUT_NANOS);
 368
 369         // No problem
 370         final boolean ret = queue.runTimeout();
 371         assertFalse(ret);
 372     }
 373
 374     @Test
 375     public void testRunTimeoutEmptyWithoutProgressMore() throws NoProgressException {
 376         ticker.increment(SequencedQueue.NO_PROGRESS_TIMEOUT_NANOS + 1);
 377
 378         // No problem
 379         final boolean ret = queue.runTimeout();
 380         assertFalse(ret);
 381     }
 382
 383     @Test
 384     public void testCompleteEmpty() {
 385         final ClientActorBehavior ret = queue.complete(mockBehavior, mockResponse);
 386         assertSame(mockBehavior, ret);
 387         verifyNoMoreInteractions(mockCallback);
 388     }
 389
 390     @Test
 391     public void testCompleteSingle() {
 392         queue.enqueueRequest(mockRequest, mockCallback);
 393
 394         ClientActorBehavior ret = queue.complete(mockBehavior, mockResponse);
 395         verify(mockCallback).complete(mockResponse);
 396         assertSame(mockBehavior, ret);
 397
 398         ret = queue.complete(mockBehavior, mockResponse);
 399         assertSame(mockBehavior, ret);
 400         verifyNoMoreInteractions(mockCallback);
 401     }
 402
 403     @Test
 404     public void testCompleteNull() {
 405         queue.enqueueRequest(mockRequest, mockCallback);
 406
 407         doReturn(null).when(mockCallback).complete(mockResponse);
 408
 409         ClientActorBehavior ret = queue.complete(mockBehavior, mockResponse);
 410         verify(mockCallback).complete(mockResponse);
 411         assertNull(ret);
 412     }
 413
 414     @Test
 415     public void testProgressRecord() throws NoProgressException {
 416         setupBackend();
 417
 418         queue.enqueueRequest(mockRequest, mockCallback);
 419
 420         ticker.increment(10);
 421         queue.enqueueRequest(mockRequest2, mockCallback);
 422         queue.complete(mockBehavior, mockResponse);
 423
 424         ticker.increment(SequencedQueue.NO_PROGRESS_TIMEOUT_NANOS - 11);
 425         assertTrue(queue.runTimeout());
 426     }
 427
 428     private void setupBackend() {
 429         final CompletableFuture<BackendInfo> proof = new CompletableFuture<>();
 430         assertTrue(queue.expectProof(proof));
 431         final Optional<FiniteDuration> ret = queue.setBackendInfo(proof, mockBackendInfo);
 432         assertNotNull(ret);
 433         assertFalse(ret.isPresent());
 434         assertFalse(mockActor.msgAvailable());
 435     }
 436
 437     private void assertTransmit(final Request<?, ?> expected) {
 438         assertTrue(mockActor.msgAvailable());
 439         assertRequestEquals(expected, mockActor.receiveOne(FiniteDuration.apply(5, TimeUnit.SECONDS)));
 440     }
 441
 442     private static void assertRequestEquals(final Request<?, ?> expected, final Object o) {
 443         final Request<?, ?> actual = (Request<?, ?>) o;
 444         assertEquals(expected.getRetry(), actual.getRetry());
 445         assertEquals(expected.getSequence(), actual.getSequence());
 446         assertEquals(expected.getTarget(), actual.getTarget());
 447     }
 448 }