2 * Copyright (c) 2017 Pantheon Technologies s.r.o. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.access.client;
10 import static org.hamcrest.CoreMatchers.everyItem;
11 import static org.mockito.Matchers.any;
12 import static org.mockito.Mockito.mock;
13 import static org.mockito.Mockito.verify;
14 import static org.opendaylight.controller.cluster.access.client.ConnectionEntryMatcher.entryWithRequest;
16 import akka.actor.ActorSystem;
17 import akka.testkit.TestProbe;
18 import akka.testkit.javadsl.TestKit;
19 import com.google.common.base.Ticker;
20 import java.util.Collection;
21 import java.util.Optional;
22 import java.util.function.Consumer;
23 import org.junit.After;
24 import org.junit.Assert;
25 import org.junit.Before;
26 import org.junit.Test;
27 import org.opendaylight.controller.cluster.access.commands.TransactionFailure;
28 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
29 import org.opendaylight.controller.cluster.access.commands.TransactionPurgeResponse;
30 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
31 import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
32 import org.opendaylight.controller.cluster.access.concepts.FrontendType;
33 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
34 import org.opendaylight.controller.cluster.access.concepts.MemberName;
35 import org.opendaylight.controller.cluster.access.concepts.Request;
36 import org.opendaylight.controller.cluster.access.concepts.RequestSuccess;
37 import org.opendaylight.controller.cluster.access.concepts.Response;
38 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
39 import org.opendaylight.controller.cluster.access.concepts.SuccessEnvelope;
40 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
42 public abstract class AbstractTransmitQueueTest<T extends TransmitQueue> {
44 private static final FrontendIdentifier FRONTEND =
45 FrontendIdentifier.create(MemberName.forName("test"), FrontendType.forName("type-1"));
46 private static final ClientIdentifier CLIENT = ClientIdentifier.create(FRONTEND, 0);
47 protected static final LocalHistoryIdentifier HISTORY = new LocalHistoryIdentifier(CLIENT, 0);
48 protected static final TransactionIdentifier TRANSACTION_IDENTIFIER = new TransactionIdentifier(HISTORY, 0);
50 protected ActorSystem system;
51 protected TestProbe probe;
53 protected abstract int getMaxInFlightMessages();
55 protected abstract T createQueue();
58 public void setUp() throws Exception {
59 system = ActorSystem.apply();
60 probe = new TestProbe(system);
61 queue = createQueue();
65 public void tearDown() throws Exception {
66 TestKit.shutdownActorSystem(system);
70 public abstract void testCanTransmitCount() throws Exception;
72 @Test(expected = UnsupportedOperationException.class)
73 public abstract void testTransmit() throws Exception;
76 public void testAsIterable() throws Exception {
77 final Request<?, ?> request = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 0L, probe.ref());
78 final Consumer<Response<?, ?>> callback = createConsumerMock();
79 final long now = Ticker.systemTicker().read();
80 final int sentMessages = getMaxInFlightMessages() + 1;
81 for (int i = 0; i < sentMessages; i++) {
82 queue.enqueueOrForward(new ConnectionEntry(request, callback, now), now);
84 final Collection<ConnectionEntry> entries = queue.drain();
85 Assert.assertEquals(sentMessages, entries.size());
86 Assert.assertThat(entries, everyItem(entryWithRequest(request)));
90 public void testTicksStalling() throws Exception {
91 final long now = Ticker.systemTicker().read();
92 Assert.assertEquals(0, queue.ticksStalling(now));
96 public void testCompleteReponseNotMatchingRequest() throws Exception {
97 final long requestSequence = 0L;
98 final long txSequence = 0L;
99 final long sessionId = 0L;
100 final Request<?, ?> request = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, requestSequence, probe.ref());
101 final Consumer<Response<?, ?>> callback = createConsumerMock();
102 final long now = Ticker.systemTicker().read();
103 queue.enqueueOrForward(new ConnectionEntry(request, callback, now), now);
104 //different transaction id
105 final TransactionIdentifier anotherTxId = new TransactionIdentifier(HISTORY, 1L);
106 final RequestSuccess<?, ?> success1 = new TransactionPurgeResponse(anotherTxId, requestSequence);
107 final Optional<TransmittedConnectionEntry> completed1 =
108 queue.complete(new SuccessEnvelope(success1, sessionId, txSequence, 1L), now);
109 Assert.assertFalse(completed1.isPresent());
110 //different response sequence
111 final long differentResponseSequence = 1L;
112 final RequestSuccess<?, ?> success2 =
113 new TransactionPurgeResponse(TRANSACTION_IDENTIFIER, differentResponseSequence);
114 final Optional<TransmittedConnectionEntry> completed2 =
115 queue.complete(new SuccessEnvelope(success2, sessionId, txSequence, 1L), now);
116 Assert.assertFalse(completed2.isPresent());
117 //different tx sequence
118 final long differentTxSequence = 1L;
119 final RequestSuccess<?, ?> success3 =
120 new TransactionPurgeResponse(TRANSACTION_IDENTIFIER, requestSequence);
121 final Optional<TransmittedConnectionEntry> completed3 =
122 queue.complete(new SuccessEnvelope(success3, sessionId, differentTxSequence, 1L), now);
123 Assert.assertFalse(completed3.isPresent());
124 //different session id
125 final long differentSessionId = 1L;
126 final RequestSuccess<?, ?> success4 =
127 new TransactionPurgeResponse(TRANSACTION_IDENTIFIER, requestSequence);
128 final Optional<TransmittedConnectionEntry> completed4 =
129 queue.complete(new SuccessEnvelope(success4, differentSessionId, differentTxSequence, 1L), now);
130 Assert.assertFalse(completed4.isPresent());
134 public void testIsEmpty() throws Exception {
135 Assert.assertTrue(queue.isEmpty());
136 final Request<?, ?> request = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 0L, probe.ref());
137 final Consumer<Response<?, ?>> callback = createConsumerMock();
138 final long now = Ticker.systemTicker().read();
139 queue.enqueueOrForward(new ConnectionEntry(request, callback, now), now);
140 Assert.assertFalse(queue.isEmpty());
144 public void testPeek() throws Exception {
145 final Request<?, ?> request1 = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 0L, probe.ref());
146 final Request<?, ?> request2 = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 1L, probe.ref());
147 final Consumer<Response<?, ?>> callback = createConsumerMock();
148 final long now = Ticker.systemTicker().read();
149 final ConnectionEntry entry1 = new ConnectionEntry(request1, callback, now);
150 final ConnectionEntry entry2 = new ConnectionEntry(request2, callback, now);
151 queue.enqueueOrForward(entry1, now);
152 queue.enqueueOrForward(entry2, now);
153 Assert.assertEquals(entry1.getRequest(), queue.peek().getRequest());
157 public void testPoison() throws Exception {
158 final Request<?, ?> request = new TransactionPurgeRequest(TRANSACTION_IDENTIFIER, 0L, probe.ref());
159 final Consumer<Response<?, ?>> callback = createConsumerMock();
160 final long now = Ticker.systemTicker().read();
161 queue.enqueueOrForward(new ConnectionEntry(request, callback, now), now);
162 queue.poison(new RuntimeRequestException("fail", new RuntimeException("fail")));
163 verify(callback).accept(any(TransactionFailure.class));
164 Assert.assertTrue(queue.isEmpty());
167 @SuppressWarnings("unchecked")
168 protected static Consumer<Response<?, ?>> createConsumerMock() {
169 return mock(Consumer.class);