2 * Copyright (c) 2014 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.openflowjava.protocol.impl.core.connection;
10 import io.netty.channel.ChannelHandler;
11 import io.netty.channel.ChannelHandlerContext;
12 import io.netty.channel.ChannelOutboundHandlerAdapter;
13 import io.netty.channel.ChannelPromise;
14 import io.netty.channel.embedded.EmbeddedChannel;
16 import java.net.InetSocketAddress;
17 import java.util.concurrent.TimeUnit;
19 import org.junit.After;
20 import org.junit.Assert;
21 import org.junit.Before;
22 import org.junit.Test;
23 import org.mockito.Mock;
24 import org.mockito.MockitoAnnotations;
25 import org.opendaylight.openflowjava.protocol.impl.core.connection.ChannelOutboundQueue;
26 import org.opendaylight.openflowjava.protocol.impl.core.connection.ConnectionAdapterImpl;
27 import org.opendaylight.openflowjava.protocol.impl.core.connection.MessageListenerWrapper;
28 import org.opendaylight.openflowjava.protocol.impl.core.connection.ResponseExpectedRpcListener;
29 import org.opendaylight.openflowjava.protocol.impl.core.connection.RpcResponseKey;
30 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoInput;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoReplyInput;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterInput;
35 import com.google.common.cache.Cache;
36 import com.google.common.cache.CacheBuilder;
37 import com.google.common.cache.RemovalListener;
38 import com.google.common.cache.RemovalNotification;
44 public class ChannelOutboundQueue02Test {
45 private static int counter;
46 private static final int RPC_RESPONSE_EXPIRATION = 1;
47 private static final RemovalListener<RpcResponseKey, ResponseExpectedRpcListener<?>> REMOVAL_LISTENER =
48 new RemovalListener<RpcResponseKey, ResponseExpectedRpcListener<?>>() {
50 public void onRemoval(
51 final RemovalNotification<RpcResponseKey, ResponseExpectedRpcListener<?>> notification) {
52 notification.getValue().discard();
56 @Mock EchoInput echoInput;
57 @Mock BarrierInput barrierInput;
58 @Mock EchoReplyInput echoReplyInput;
59 @Mock ExperimenterInput experimenterInput;
60 private ConnectionAdapterImpl adapter;
61 private Cache<RpcResponseKey, ResponseExpectedRpcListener<?>> cache;
67 MockitoAnnotations.initMocks(this);
70 * Disconnect adapter after each test
73 public void tierDown(){
74 if (adapter != null && adapter.isAlive()) {
80 * Test write to closed / opened channel
84 public void test01() throws Exception {
85 EmbeddedChannel ec = new EmbeddedChannel(new EmbededChannelHandler());
86 adapter = new ConnectionAdapterImpl(ec,InetSocketAddress.createUnresolved("localhost", 9876));
87 cache = CacheBuilder.newBuilder().concurrencyLevel(1).expireAfterWrite(RPC_RESPONSE_EXPIRATION, TimeUnit.MINUTES)
88 .removalListener(REMOVAL_LISTENER).build();
89 adapter.setResponseCache(cache);
90 ChannelOutboundQueue cq = (ChannelOutboundQueue) ec.pipeline().last();
92 adapter.barrier(barrierInput);
93 adapter.echo(echoInput);
94 cq.channelInactive(ec.pipeline().lastContext());
96 Assert.assertEquals("Wrong - ChannelOutboundHandlerAdapter.write was invoked on closed channel",0, counter);
97 cq.channelActive(ec.pipeline().lastContext());
99 adapter.barrier(barrierInput);
100 adapter.experimenter(experimenterInput);
101 ec.runPendingTasks();
102 Assert.assertEquals("Wrong - ChannelOutboundHandlerAdapter.write has not been invoked on opened channel",2, counter);
106 * Test write to read only / writable channel
109 public void test02(){
110 ChangeWritableEmbededChannel ec = new ChangeWritableEmbededChannel(new EmbededChannelHandler());
111 adapter = new ConnectionAdapterImpl(ec,InetSocketAddress.createUnresolved("localhost", 9876));
112 cache = CacheBuilder.newBuilder().concurrencyLevel(1).expireAfterWrite(RPC_RESPONSE_EXPIRATION, TimeUnit.MINUTES)
113 .removalListener(REMOVAL_LISTENER).build();
114 adapter.setResponseCache(cache);
117 adapter.barrier(barrierInput);
118 adapter.echo(echoInput);
119 ec.runPendingTasks();
120 Assert.assertEquals("Wrong - write to readonly channel",0, counter);
122 adapter.echoReply(echoReplyInput);
123 adapter.echo(echoInput);
124 ec.runPendingTasks();
125 Assert.assertEquals("Wrong - write to writtable channel",4, counter);
129 * Channel Handler for testing
133 private class EmbededChannelHandler extends ChannelOutboundHandlerAdapter {
135 public void write(ChannelHandlerContext ctx, Object msg,
136 ChannelPromise promise) throws Exception {
137 if(msg instanceof MessageListenerWrapper){
144 * Class for testing - channel can change state to read only or writable
148 private class ChangeWritableEmbededChannel extends EmbeddedChannel {
149 private boolean isWrittable;
150 public ChangeWritableEmbededChannel(ChannelHandler channelHandler){
151 super(channelHandler);
156 public boolean isWritable() {
160 public void setWritable(){
164 public void setReadOnly(){