opendaylight/md-sal/sal-clustering-commons/src/test/java/org/opendaylight/controller/cluster/messaging/MessageSlicerTest.java

   1 /*
   2  * Copyright (c) 2017 Inocybe Technologies 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.messaging;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.mockito.Matchers.any;
  12 import static org.mockito.Matchers.anyInt;
  13 import static org.mockito.Mockito.doNothing;
  14 import static org.mockito.Mockito.doReturn;
  15 import static org.mockito.Mockito.doThrow;
  16 import static org.mockito.Mockito.verify;
  17 import static org.mockito.Mockito.verifyNoMoreInteractions;
  18
  19 import akka.actor.ActorRef;
  20 import com.google.common.util.concurrent.Uninterruptibles;
  21 import java.io.IOException;
  22 import java.io.Serializable;
  23 import java.util.concurrent.TimeUnit;
  24 import java.util.function.Consumer;
  25 import org.junit.Before;
  26 import org.junit.Test;
  27 import org.mockito.ArgumentCaptor;
  28 import org.mockito.Mock;
  29 import org.opendaylight.yangtools.concepts.Identifier;
  30
  31 /**
  32  * Unit tests for MessageSlicer.
  33  *
  34  * @author Thomas Pantelis
  35  */
  36 public class MessageSlicerTest extends AbstractMessagingTest {
  37     @Mock
  38     private Consumer<Throwable> mockOnFailureCallback;
  39
  40     @Override
  41     @Before
  42     public void setup() throws IOException {
  43         super.setup();
  44
  45         doNothing().when(mockOnFailureCallback).accept(any(Throwable.class));
  46     }
  47
  48     @Test
  49     public void testHandledMessages() {
  50         final MessageSliceReply reply = MessageSliceReply.success(IDENTIFIER, 1, testProbe.ref());
  51         assertEquals("isHandledMessage", Boolean.TRUE, MessageSlicer.isHandledMessage(reply));
  52         assertEquals("isHandledMessage", Boolean.FALSE, MessageSlicer.isHandledMessage(new Object()));
  53
  54         try (MessageSlicer slicer = newMessageSlicer("testHandledMessages", 100)) {
  55             assertEquals("handledMessage", Boolean.TRUE, slicer.handleMessage(reply));
  56             assertEquals("handledMessage", Boolean.FALSE, slicer.handleMessage(new Object()));
  57         }
  58     }
  59
  60     @Test
  61     public void testSliceWithFailedSerialization() throws IOException {
  62         IOException mockFailure = new IOException("mock IOException");
  63         doThrow(mockFailure).when(mockFiledBackedStream).write(any(byte[].class), anyInt(), anyInt());
  64         doThrow(mockFailure).when(mockFiledBackedStream).write(any(byte[].class));
  65         doThrow(mockFailure).when(mockFiledBackedStream).write(anyInt());
  66         doThrow(mockFailure).when(mockFiledBackedStream).flush();
  67
  68         try (MessageSlicer slicer = newMessageSlicer("testSliceWithFailedSerialization", 100)) {
  69             slice(slicer, IDENTIFIER, new BytesMessage(new byte[]{}), testProbe.ref(), testProbe.ref(),
  70                     mockOnFailureCallback);
  71
  72             assertFailureCallback(IOException.class);
  73             verify(mockFiledBackedStream).cleanup();
  74         }
  75     }
  76
  77     @Test
  78     public void testSliceWithByteSourceFailure() throws IOException {
  79         IOException mockFailure = new IOException("mock IOException");
  80         doThrow(mockFailure).when(mockByteSource).openStream();
  81         doThrow(mockFailure).when(mockByteSource).openBufferedStream();
  82
  83         try (MessageSlicer slicer = newMessageSlicer("testSliceWithByteSourceFailure", 100)) {
  84             slice(slicer, IDENTIFIER, new BytesMessage(new byte[]{}), testProbe.ref(), testProbe.ref(),
  85                     mockOnFailureCallback);
  86
  87             assertFailureCallback(IOException.class);
  88             verify(mockFiledBackedStream).cleanup();
  89         }
  90     }
  91
  92     @Test
  93     public void testSliceWithInputStreamFailure() throws IOException {
  94         doReturn(0).when(mockInputStream).read(any(byte[].class));
  95
  96         try (MessageSlicer slicer = newMessageSlicer("testSliceWithInputStreamFailure", 2)) {
  97             slice(slicer, IDENTIFIER, new BytesMessage(new byte[]{}), testProbe.ref(), testProbe.ref(),
  98                     mockOnFailureCallback);
  99
 100             assertFailureCallback(IOException.class);
 101             verify(mockFiledBackedStream).cleanup();
 102         }
 103     }
 104
 105     @Test
 106     public void testMessageSliceReplyWithNoState() {
 107         try (MessageSlicer slicer = newMessageSlicer("testMessageSliceReplyWithNoState", 1000)) {
 108             slicer.handleMessage(MessageSliceReply.success(IDENTIFIER, 1, testProbe.ref()));
 109             final AbortSlicing abortSlicing = testProbe.expectMsgClass(AbortSlicing.class);
 110             assertEquals("Identifier", IDENTIFIER, abortSlicing.getIdentifier());
 111         }
 112     }
 113
 114     @Test
 115     public void testCloseAllSlicedMessageState() throws IOException {
 116         doReturn(1).when(mockInputStream).read(any(byte[].class));
 117
 118         final MessageSlicer slicer = newMessageSlicer("testCloseAllSlicedMessageState", 1);
 119         slice(slicer, IDENTIFIER, new BytesMessage(new byte[]{1, 2}), testProbe.ref(), testProbe.ref(),
 120                 mockOnFailureCallback);
 121
 122         slicer.close();
 123
 124         verify(mockFiledBackedStream).cleanup();
 125         verifyNoMoreInteractions(mockOnFailureCallback);
 126     }
 127
 128     @Test
 129     public void testCheckExpiredSlicedMessageState() throws IOException {
 130         doReturn(1).when(mockInputStream).read(any(byte[].class));
 131
 132         final int expiryDuration = 200;
 133         try (MessageSlicer slicer = MessageSlicer.builder().messageSliceSize(1)
 134                 .logContext("testCheckExpiredSlicedMessageState")
 135                 .filedBackedStreamFactory(mockFiledBackedStreamFactory)
 136                 .expireStateAfterInactivity(expiryDuration, TimeUnit.MILLISECONDS).build()) {
 137             slice(slicer, IDENTIFIER, new BytesMessage(new byte[]{1, 2}), testProbe.ref(), testProbe.ref(),
 138                     mockOnFailureCallback);
 139
 140             Uninterruptibles.sleepUninterruptibly(expiryDuration + 50, TimeUnit.MILLISECONDS);
 141             slicer.checkExpiredSlicedMessageState();
 142
 143             assertFailureCallback(RuntimeException.class);
 144             verify(mockFiledBackedStream).cleanup();
 145         }
 146     }
 147
 148     private void assertFailureCallback(final Class<?> exceptionType) {
 149         ArgumentCaptor<Throwable> exceptionCaptor = ArgumentCaptor.forClass(Throwable.class);
 150         verify(mockOnFailureCallback).accept(exceptionCaptor.capture());
 151         assertEquals("Exception type", exceptionType, exceptionCaptor.getValue().getClass());
 152     }
 153
 154     private MessageSlicer newMessageSlicer(String logContext, final int messageSliceSize) {
 155         return MessageSlicer.builder().messageSliceSize(messageSliceSize).logContext(logContext)
 156                 .filedBackedStreamFactory(mockFiledBackedStreamFactory).build();
 157     }
 158
 159     static void slice(MessageSlicer slicer, Identifier identifier, Serializable message, ActorRef sendTo,
 160             ActorRef replyTo, Consumer<Throwable> onFailureCallback) {
 161         slicer.slice(SliceOptions.builder().identifier(identifier).message(message).sendTo(sendTo).replyTo(replyTo)
 162                 .onFailureCallback(onFailureCallback).build());
 163     }
 164 }