홍길동 / onos

Go to a project
Toggle navigation Toggle navigation pinning
Thomas Vachuska
Committed by Gerrit Code Review
b093c916 1 parent 6e0fb302

Renamed IOLoop & Netty to *MessagingManager for consistency. 

Change-Id: Id8859e24d0c7ac7f948516388069639093bad524
Inline Side-by-side
Showing 6 changed files with 27 additions and 24 deletions
...@@ -25,7 +25,7 @@ import org.apache.felix.scr.annotations.Component; ...@@ -25,7 +25,7 @@ import org.apache.felix.scr.annotations.Component;
25 import org.apache.felix.scr.annotations.Deactivate; 25 import org.apache.felix.scr.annotations.Deactivate;
26 import org.apache.felix.scr.annotations.Service; 26 import org.apache.felix.scr.annotations.Service;
27 import org.joda.time.DateTime; 27 import org.joda.time.DateTime;
28 -import org.onlab.netty.NettyMessagingService; 28 +import org.onlab.netty.NettyMessagingManager;
29 import org.onlab.packet.IpAddress; 29 import org.onlab.packet.IpAddress;
30 import org.onlab.util.KryoNamespace; 30 import org.onlab.util.KryoNamespace;
31 import org.onosproject.cluster.ClusterEvent; 31 import org.onosproject.cluster.ClusterEvent;
...@@ -108,7 +108,7 @@ public class DistributedClusterStore ...@@ -108,7 +108,7 @@ public class DistributedClusterStore
108 private final Map<NodeId, ControllerNode> allNodes = Maps.newConcurrentMap(); 108 private final Map<NodeId, ControllerNode> allNodes = Maps.newConcurrentMap();
109 private final Map<NodeId, State> nodeStates = Maps.newConcurrentMap(); 109 private final Map<NodeId, State> nodeStates = Maps.newConcurrentMap();
110 private final Map<NodeId, DateTime> nodeStateLastUpdatedTimes = Maps.newConcurrentMap(); 110 private final Map<NodeId, DateTime> nodeStateLastUpdatedTimes = Maps.newConcurrentMap();
111 - private NettyMessagingService messagingService; 111 + private NettyMessagingManager messagingService;
112 private ScheduledExecutorService heartBeatSender = Executors.newSingleThreadScheduledExecutor( 112 private ScheduledExecutorService heartBeatSender = Executors.newSingleThreadScheduledExecutor(
113 groupedThreads("onos/cluster/membership", "heartbeat-sender")); 113 groupedThreads("onos/cluster/membership", "heartbeat-sender"));
114 private ExecutorService heartBeatMessageHandler = Executors.newSingleThreadExecutor( 114 private ExecutorService heartBeatMessageHandler = Executors.newSingleThreadExecutor(
...@@ -148,7 +148,7 @@ public class DistributedClusterStore ...@@ -148,7 +148,7 @@ public class DistributedClusterStore
148 148
149 establishSelfIdentity(); 149 establishSelfIdentity();
150 150
151 - messagingService = new NettyMessagingService(HEARTBEAT_FD_PORT); 151 + messagingService = new NettyMessagingManager(HEARTBEAT_FD_PORT);
152 try { 152 try {
153 messagingService.activate(); 153 messagingService.activate();
154 } catch (InterruptedException e) { 154 } catch (InterruptedException e) {
......
...@@ -21,8 +21,8 @@ import org.apache.felix.scr.annotations.Deactivate; ...@@ -21,8 +21,8 @@ import org.apache.felix.scr.annotations.Deactivate;
21 import org.apache.felix.scr.annotations.Reference; 21 import org.apache.felix.scr.annotations.Reference;
22 import org.apache.felix.scr.annotations.ReferenceCardinality; 22 import org.apache.felix.scr.annotations.ReferenceCardinality;
23 import org.apache.felix.scr.annotations.Service; 23 import org.apache.felix.scr.annotations.Service;
24 -import org.onlab.netty.NettyMessagingService; 24 +import org.onlab.netty.NettyMessagingManager;
25 -import org.onlab.nio.service.IOLoopMessagingService; 25 +import org.onlab.nio.service.IOLoopMessagingManager;
26 import org.onosproject.cluster.ClusterService; 26 import org.onosproject.cluster.ClusterService;
27 import org.onosproject.cluster.ControllerNode; 27 import org.onosproject.cluster.ControllerNode;
28 import org.onosproject.cluster.NodeId; 28 import org.onosproject.cluster.NodeId;
...@@ -69,7 +69,7 @@ public class ClusterCommunicationManager ...@@ -69,7 +69,7 @@ public class ClusterCommunicationManager
69 public void activate() { 69 public void activate() {
70 ControllerNode localNode = clusterService.getLocalNode(); 70 ControllerNode localNode = clusterService.getLocalNode();
71 if (useNetty) { 71 if (useNetty) {
72 - NettyMessagingService netty = new NettyMessagingService(localNode.ip(), localNode.tcpPort()); 72 + NettyMessagingManager netty = new NettyMessagingManager(localNode.ip(), localNode.tcpPort());
73 try { 73 try {
74 netty.activate(); 74 netty.activate();
75 messagingService = netty; 75 messagingService = netty;
...@@ -77,7 +77,7 @@ public class ClusterCommunicationManager ...@@ -77,7 +77,7 @@ public class ClusterCommunicationManager
77 log.error("NettyMessagingService#activate", e); 77 log.error("NettyMessagingService#activate", e);
78 } 78 }
79 } else { 79 } else {
80 - IOLoopMessagingService ioLoop = new IOLoopMessagingService(localNode.ip(), localNode.tcpPort()); 80 + IOLoopMessagingManager ioLoop = new IOLoopMessagingManager(localNode.ip(), localNode.tcpPort());
81 try { 81 try {
82 ioLoop.activate(); 82 ioLoop.activate();
83 messagingService = ioLoop; 83 messagingService = ioLoop;
...@@ -94,9 +94,9 @@ public class ClusterCommunicationManager ...@@ -94,9 +94,9 @@ public class ClusterCommunicationManager
94 // FIXME: workaround until it becomes a service. 94 // FIXME: workaround until it becomes a service.
95 try { 95 try {
96 if (useNetty) { 96 if (useNetty) {
97 - ((NettyMessagingService) messagingService).deactivate(); 97 + ((NettyMessagingManager) messagingService).deactivate();
98 } else { 98 } else {
99 - ((IOLoopMessagingService) messagingService).deactivate(); 99 + ((IOLoopMessagingManager) messagingService).deactivate();
100 } 100 }
101 } catch (Exception e) { 101 } catch (Exception e) {
102 log.error("MessagingService#deactivate", e); 102 log.error("MessagingService#deactivate", e);
......
...@@ -22,7 +22,7 @@ import org.junit.Test; ...@@ -22,7 +22,7 @@ import org.junit.Test;
22 import org.onosproject.cluster.DefaultControllerNode; 22 import org.onosproject.cluster.DefaultControllerNode;
23 import org.onosproject.cluster.NodeId; 23 import org.onosproject.cluster.NodeId;
24 import org.onosproject.store.cluster.impl.ClusterNodesDelegate; 24 import org.onosproject.store.cluster.impl.ClusterNodesDelegate;
25 -import org.onlab.netty.NettyMessagingService; 25 +import org.onlab.netty.NettyMessagingManager;
26 import org.onlab.packet.IpAddress; 26 import org.onlab.packet.IpAddress;
27 27
28 import java.util.concurrent.CountDownLatch; 28 import java.util.concurrent.CountDownLatch;
...@@ -56,7 +56,7 @@ public class ClusterCommunicationManagerTest { ...@@ -56,7 +56,7 @@ public class ClusterCommunicationManagerTest {
56 @Before 56 @Before
57 public void setUp() throws Exception { 57 public void setUp() throws Exception {
58 58
59 - NettyMessagingService messagingService = new NettyMessagingService(); 59 + NettyMessagingManager messagingService = new NettyMessagingManager();
60 messagingService.activate(); 60 messagingService.activate();
61 61
62 ccm1 = new ClusterCommunicationManager(); 62 ccm1 = new ClusterCommunicationManager();
......
...@@ -65,7 +65,7 @@ import com.google.common.cache.RemovalNotification; ...@@ -65,7 +65,7 @@ import com.google.common.cache.RemovalNotification;
65 /** 65 /**
66 * Implementation of MessagingService based on <a href="http://netty.io/">Netty</a> framework. 66 * Implementation of MessagingService based on <a href="http://netty.io/">Netty</a> framework.
67 */ 67 */
68 -public class NettyMessagingService implements MessagingService { 68 +public class NettyMessagingManager implements MessagingService {
69 69
70 private final Logger log = LoggerFactory.getLogger(getClass()); 70 private final Logger log = LoggerFactory.getLogger(getClass());
71 71
...@@ -111,15 +111,15 @@ public class NettyMessagingService implements MessagingService { ...@@ -111,15 +111,15 @@ public class NettyMessagingService implements MessagingService {
111 clientChannelClass = NioSocketChannel.class; 111 clientChannelClass = NioSocketChannel.class;
112 } 112 }
113 113
114 - public NettyMessagingService(IpAddress ip, int port) { 114 + public NettyMessagingManager(IpAddress ip, int port) {
115 localEp = new Endpoint(ip, port); 115 localEp = new Endpoint(ip, port);
116 } 116 }
117 117
118 - public NettyMessagingService() { 118 + public NettyMessagingManager() {
119 this(8080); 119 this(8080);
120 } 120 }
121 121
122 - public NettyMessagingService(int port) { 122 + public NettyMessagingManager(int port) {
123 try { 123 try {
124 localEp = new Endpoint(IpAddress.valueOf(InetAddress.getLocalHost()), port); 124 localEp = new Endpoint(IpAddress.valueOf(InetAddress.getLocalHost()), port);
125 } catch (UnknownHostException e) { 125 } catch (UnknownHostException e) {
......
...@@ -38,8 +38,8 @@ public class PingPongTest { ...@@ -38,8 +38,8 @@ public class PingPongTest {
38 @Ignore("Turning off fragile test") 38 @Ignore("Turning off fragile test")
39 @Test 39 @Test
40 public void testPingPong() throws Exception { 40 public void testPingPong() throws Exception {
41 - NettyMessagingService pinger = new NettyMessagingService(8085); 41 + NettyMessagingManager pinger = new NettyMessagingManager(8085);
42 - NettyMessagingService ponger = new NettyMessagingService(9086); 42 + NettyMessagingManager ponger = new NettyMessagingManager(9086);
43 try { 43 try {
44 pinger.activate(); 44 pinger.activate();
45 ponger.activate(); 45 ponger.activate();
......
...@@ -38,6 +38,7 @@ import java.util.function.Function; ...@@ -38,6 +38,7 @@ import java.util.function.Function;
38 import org.apache.commons.pool.KeyedPoolableObjectFactory; 38 import org.apache.commons.pool.KeyedPoolableObjectFactory;
39 import org.apache.commons.pool.impl.GenericKeyedObjectPool; 39 import org.apache.commons.pool.impl.GenericKeyedObjectPool;
40 import org.onlab.nio.AcceptorLoop; 40 import org.onlab.nio.AcceptorLoop;
41 +import org.onlab.nio.SelectorLoop;
41 import org.onlab.packet.IpAddress; 42 import org.onlab.packet.IpAddress;
42 import org.onosproject.store.cluster.messaging.Endpoint; 43 import org.onosproject.store.cluster.messaging.Endpoint;
43 import org.onosproject.store.cluster.messaging.MessagingService; 44 import org.onosproject.store.cluster.messaging.MessagingService;
...@@ -53,7 +54,7 @@ import com.google.common.collect.Lists; ...@@ -53,7 +54,7 @@ import com.google.common.collect.Lists;
53 /** 54 /**
54 * MessagingService implementation based on IOLoop. 55 * MessagingService implementation based on IOLoop.
55 */ 56 */
56 -public class IOLoopMessagingService implements MessagingService { 57 +public class IOLoopMessagingManager implements MessagingService {
57 58
58 private final Logger log = LoggerFactory.getLogger(getClass()); 59 private final Logger log = LoggerFactory.getLogger(getClass());
59 60
...@@ -81,7 +82,7 @@ public class IOLoopMessagingService implements MessagingService { ...@@ -81,7 +82,7 @@ public class IOLoopMessagingService implements MessagingService {
81 private final Endpoint localEp; 82 private final Endpoint localEp;
82 83
83 private GenericKeyedObjectPool<Endpoint, DefaultMessageStream> streams = 84 private GenericKeyedObjectPool<Endpoint, DefaultMessageStream> streams =
84 - new GenericKeyedObjectPool<Endpoint, DefaultMessageStream>(new DefaultMessageStreamFactory()); 85 + new GenericKeyedObjectPool<>(new DefaultMessageStreamFactory());
85 86
86 private final ConcurrentMap<String, Consumer<DefaultMessage>> handlers = new ConcurrentHashMap<>(); 87 private final ConcurrentMap<String, Consumer<DefaultMessage>> handlers = new ConcurrentHashMap<>();
87 private final AtomicLong messageIdGenerator = new AtomicLong(0); 88 private final AtomicLong messageIdGenerator = new AtomicLong(0);
...@@ -97,20 +98,21 @@ public class IOLoopMessagingService implements MessagingService { ...@@ -97,20 +98,21 @@ public class IOLoopMessagingService implements MessagingService {
97 .build(); 98 .build();
98 99
99 100
100 - public IOLoopMessagingService(int port) { 101 + public IOLoopMessagingManager(int port) {
101 this(new Endpoint(IpAddress.valueOf("127.0.0.1"), port)); 102 this(new Endpoint(IpAddress.valueOf("127.0.0.1"), port));
102 } 103 }
103 104
104 - public IOLoopMessagingService(IpAddress ip, int port) { 105 + public IOLoopMessagingManager(IpAddress ip, int port) {
105 this(new Endpoint(ip, port)); 106 this(new Endpoint(ip, port));
106 } 107 }
107 108
108 - public IOLoopMessagingService(Endpoint localEp) { 109 + public IOLoopMessagingManager(Endpoint localEp) {
109 this.localEp = localEp; 110 this.localEp = localEp;
110 } 111 }
111 112
112 /** 113 /**
113 * Returns the local endpoint. 114 * Returns the local endpoint.
115 + *
114 * @return local endpoint 116 * @return local endpoint
115 */ 117 */
116 public Endpoint localEp() { 118 public Endpoint localEp() {
...@@ -119,6 +121,7 @@ public class IOLoopMessagingService implements MessagingService { ...@@ -119,6 +121,7 @@ public class IOLoopMessagingService implements MessagingService {
119 121
120 /** 122 /**
121 * Activates IO Loops. 123 * Activates IO Loops.
124 + *
122 * @throws IOException is activation fails 125 * @throws IOException is activation fails
123 */ 126 */
124 public void activate() throws IOException { 127 public void activate() throws IOException {
...@@ -129,7 +132,7 @@ public class IOLoopMessagingService implements MessagingService { ...@@ -129,7 +132,7 @@ public class IOLoopMessagingService implements MessagingService {
129 ioLoops.add(new DefaultIOLoop(this::dispatchLocally)); 132 ioLoops.add(new DefaultIOLoop(this::dispatchLocally));
130 } 133 }
131 134
132 - ioLoops.forEach(loop -> ioThreadPool.execute(loop)); 135 + ioLoops.forEach(ioThreadPool::execute);
133 acceptorThreadPool.execute(acceptorLoop); 136 acceptorThreadPool.execute(acceptorLoop);
134 ioLoops.forEach(loop -> loop.awaitStart(TIMEOUT)); 137 ioLoops.forEach(loop -> loop.awaitStart(TIMEOUT));
135 acceptorLoop.awaitStart(TIMEOUT); 138 acceptorLoop.awaitStart(TIMEOUT);
...@@ -139,7 +142,7 @@ public class IOLoopMessagingService implements MessagingService { ...@@ -139,7 +142,7 @@ public class IOLoopMessagingService implements MessagingService {
139 * Shuts down IO loops. 142 * Shuts down IO loops.
140 */ 143 */
141 public void deactivate() { 144 public void deactivate() {
142 - ioLoops.forEach(loop -> loop.shutdown()); 145 + ioLoops.forEach(SelectorLoop::shutdown);
143 acceptorLoop.shutdown(); 146 acceptorLoop.shutdown();
144 ioThreadPool.shutdown(); 147 ioThreadPool.shutdown();
145 acceptorThreadPool.shutdown(); 148 acceptorThreadPool.shutdown();
......