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.hasItems;
11 import static org.hamcrest.MatcherAssert.assertThat;
12 import static org.junit.Assert.assertEquals;
13 import static org.junit.Assert.assertSame;
14 import static org.junit.Assert.assertTrue;
15 import static org.mockito.ArgumentMatchers.isA;
16 import static org.mockito.Mockito.mock;
17 import static org.mockito.Mockito.timeout;
18 import static org.mockito.Mockito.verify;
19 import static org.opendaylight.controller.cluster.access.client.ConnectionEntryMatcher.entryWithRequest;
21 import akka.actor.ActorRef;
22 import akka.actor.ActorSystem;
23 import akka.testkit.TestProbe;
24 import akka.testkit.javadsl.TestKit;
25 import com.google.common.collect.Iterables;
26 import java.util.OptionalLong;
27 import java.util.function.Consumer;
28 import org.junit.After;
29 import org.junit.Before;
30 import org.junit.Test;
31 import org.opendaylight.controller.cluster.access.commands.AbortLocalTransactionRequest;
32 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
33 import org.opendaylight.controller.cluster.access.commands.TransactionFailure;
34 import org.opendaylight.controller.cluster.access.concepts.ClientIdentifier;
35 import org.opendaylight.controller.cluster.access.concepts.FrontendIdentifier;
36 import org.opendaylight.controller.cluster.access.concepts.FrontendType;
37 import org.opendaylight.controller.cluster.access.concepts.LocalHistoryIdentifier;
38 import org.opendaylight.controller.cluster.access.concepts.MemberName;
39 import org.opendaylight.controller.cluster.access.concepts.Request;
40 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
41 import org.opendaylight.controller.cluster.access.concepts.RequestSuccess;
42 import org.opendaylight.controller.cluster.access.concepts.Response;
43 import org.opendaylight.controller.cluster.access.concepts.ResponseEnvelope;
44 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
45 import org.opendaylight.controller.cluster.access.concepts.SuccessEnvelope;
46 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
48 public abstract class AbstractClientConnectionTest<T extends AbstractClientConnection<U>, U extends BackendInfo> {
50 protected static final MemberName MEMBER_NAME = MemberName.forName("member-1");
51 protected static final FrontendType FRONTEND_TYPE =
52 FrontendType.forName(ClientActorContextTest.class.getSimpleName());
53 protected static final FrontendIdentifier FRONTEND_ID = FrontendIdentifier.create(MEMBER_NAME, FRONTEND_TYPE);
54 protected static final ClientIdentifier CLIENT_ID = ClientIdentifier.create(FRONTEND_ID, 0);
55 protected static final String PERSISTENCE_ID = "per-1";
57 protected T connection;
58 protected ClientActorContext context;
59 protected ActorSystem system;
60 protected TestProbe backendProbe;
61 protected TestProbe contextProbe;
62 protected TestProbe replyToProbe;
66 system = ActorSystem.apply();
67 backendProbe = new TestProbe(system);
68 contextProbe = new TestProbe(system);
69 context = new ClientActorContext(contextProbe.ref(), PERSISTENCE_ID, system,
70 CLIENT_ID, AccessClientUtil.newMockClientActorConfig());
71 replyToProbe = new TestProbe(system);
72 connection = createConnection();
75 protected abstract T createConnection();
78 public void testLocalActor() {
79 assertEquals(contextProbe.ref(), connection.localActor());
83 public abstract void testReconnectConnection();
86 public void testPoison() {
87 final Consumer<Response<?, ?>> callback = mock(Consumer.class);
88 final Request<?, ?> request = createRequest(replyToProbe.ref());
89 final ConnectionEntry entry = new ConnectionEntry(request, callback, 0L);
90 connection.enqueueEntry(entry, 0L);
91 connection.poison(new RuntimeRequestException("fail", new RuntimeException("fail")));
92 verify(callback, timeout(1000)).accept(isA(TransactionFailure.class));
96 public void testSendRequestReceiveResponse() {
97 final Consumer<Response<?, ?>> callback = mock(Consumer.class);
98 final Request<?, ?> request = createRequest(replyToProbe.ref());
99 connection.sendRequest(request, callback);
100 final RequestEnvelope requestEnvelope = backendProbe.expectMsgClass(RequestEnvelope.class);
101 assertEquals(request, requestEnvelope.getMessage());
102 final LocalHistoryIdentifier historyId = new LocalHistoryIdentifier(CLIENT_ID, 0L);
103 final RequestSuccess<?, ?> message = new TransactionAbortSuccess(new TransactionIdentifier(historyId, 0L), 0L);
104 final ResponseEnvelope<?> envelope = new SuccessEnvelope(message, 0L, 0L, 0L);
105 connection.receiveResponse(envelope);
106 verify(callback, timeout(1000)).accept(isA(TransactionAbortSuccess.class));
110 public void testRun() {
111 final ClientActorBehavior<U> behavior = mock(ClientActorBehavior.class);
112 assertSame(behavior, connection.runTimer(behavior));
116 public void testCheckTimeoutEmptyQueue() {
117 assertEquals(OptionalLong.empty(), connection.checkTimeout(context.ticker().read()));
121 public void testCheckTimeout() {
122 final Consumer<Response<?, ?>> callback = mock(Consumer.class);
123 connection.sendRequest(createRequest(replyToProbe.ref()), callback);
124 final long now = context.ticker().read();
125 final OptionalLong timeout = connection.checkTimeout(now);
126 assertTrue(timeout.isPresent());
130 public void testReplay() {
131 final Consumer<Response<?, ?>> callback = mock(Consumer.class);
132 final Request<?, ?> request1 = createRequest(replyToProbe.ref());
133 final Request<?, ?> request2 = createRequest(replyToProbe.ref());
134 connection.sendRequest(request1, callback);
135 connection.sendRequest(request2, callback);
136 final Iterable<ConnectionEntry> entries = connection.startReplay();
137 assertThat(entries, hasItems(entryWithRequest(request1), entryWithRequest(request2)));
138 assertEquals(2, Iterables.size(entries));
139 Iterables.removeIf(entries, e -> true);
140 final ReconnectForwarder forwarder = mock(ReconnectForwarder.class);
141 connection.finishReplay(forwarder);
145 public void tearDown() {
146 TestKit.shutdownActorSystem(system);
149 protected Request<?, ?> createRequest(final ActorRef replyTo) {
150 final TransactionIdentifier identifier =
151 new TransactionIdentifier(new LocalHistoryIdentifier(CLIENT_ID, 0L), 0L);
152 return new AbortLocalTransactionRequest(identifier, replyTo);