Bunch of printf debugging code we probably want to remove later
Change-Id: Ibc2244f71fd7e2b6d468df80b5e29a6fa01da515
Showing
1 changed file
with
20 additions
and
5 deletions
| ... | @@ -3,11 +3,11 @@ package org.onlab.onos.store.cluster.messaging.impl; | ... | @@ -3,11 +3,11 @@ package org.onlab.onos.store.cluster.messaging.impl; |
| 3 | import static com.google.common.base.Preconditions.checkArgument; | 3 | import static com.google.common.base.Preconditions.checkArgument; |
| 4 | 4 | ||
| 5 | import java.io.IOException; | 5 | import java.io.IOException; |
| 6 | -import java.util.HashMap; | ||
| 7 | -import java.util.Map; | ||
| 8 | import java.util.Set; | 6 | import java.util.Set; |
| 9 | import java.util.Timer; | 7 | import java.util.Timer; |
| 10 | import java.util.TimerTask; | 8 | import java.util.TimerTask; |
| 9 | +import java.util.concurrent.TimeUnit; | ||
| 10 | +import java.util.concurrent.TimeoutException; | ||
| 11 | 11 | ||
| 12 | import org.apache.felix.scr.annotations.Activate; | 12 | import org.apache.felix.scr.annotations.Activate; |
| 13 | import org.apache.felix.scr.annotations.Component; | 13 | import org.apache.felix.scr.annotations.Component; |
| ... | @@ -35,6 +35,7 @@ import org.onlab.netty.Message; | ... | @@ -35,6 +35,7 @@ import org.onlab.netty.Message; |
| 35 | import org.onlab.netty.MessageHandler; | 35 | import org.onlab.netty.MessageHandler; |
| 36 | import org.onlab.netty.MessagingService; | 36 | import org.onlab.netty.MessagingService; |
| 37 | import org.onlab.netty.NettyMessagingService; | 37 | import org.onlab.netty.NettyMessagingService; |
| 38 | +import org.onlab.netty.Response; | ||
| 38 | import org.slf4j.Logger; | 39 | import org.slf4j.Logger; |
| 39 | import org.slf4j.LoggerFactory; | 40 | import org.slf4j.LoggerFactory; |
| 40 | 41 | ||
| ... | @@ -121,9 +122,12 @@ public class ClusterCommunicationManager | ... | @@ -121,9 +122,12 @@ public class ClusterCommunicationManager |
| 121 | checkArgument(node != null, "Unknown nodeId: %s", toNodeId); | 122 | checkArgument(node != null, "Unknown nodeId: %s", toNodeId); |
| 122 | Endpoint nodeEp = new Endpoint(node.ip().toString(), node.tcpPort()); | 123 | Endpoint nodeEp = new Endpoint(node.ip().toString(), node.tcpPort()); |
| 123 | try { | 124 | try { |
| 124 | - messagingService.sendAsync(nodeEp, message.subject().value(), SERIALIZER.encode(message)); | 125 | + log.info("sending..."); |
| 126 | + Response resp = messagingService.sendAndReceive(nodeEp, message.subject().value(), SERIALIZER.encode(message)); | ||
| 127 | + resp.get(1, TimeUnit.SECONDS); | ||
| 128 | + log.info("sent..."); | ||
| 125 | return true; | 129 | return true; |
| 126 | - } catch (IOException e) { | 130 | + } catch (IOException | TimeoutException e) { |
| 127 | log.error("Failed to send cluster message to nodeId: " + toNodeId, e); | 131 | log.error("Failed to send cluster message to nodeId: " + toNodeId, e); |
| 128 | } | 132 | } |
| 129 | 133 | ||
| ... | @@ -191,7 +195,8 @@ public class ClusterCommunicationManager | ... | @@ -191,7 +195,8 @@ public class ClusterCommunicationManager |
| 191 | } | 195 | } |
| 192 | } | 196 | } |
| 193 | 197 | ||
| 194 | - private static class InternalClusterMessageHandler implements MessageHandler { | 198 | + // FIXME: revert static |
| 199 | + private class InternalClusterMessageHandler implements MessageHandler { | ||
| 195 | 200 | ||
| 196 | private final ClusterMessageHandler handler; | 201 | private final ClusterMessageHandler handler; |
| 197 | 202 | ||
| ... | @@ -201,8 +206,18 @@ public class ClusterCommunicationManager | ... | @@ -201,8 +206,18 @@ public class ClusterCommunicationManager |
| 201 | 206 | ||
| 202 | @Override | 207 | @Override |
| 203 | public void handle(Message message) { | 208 | public void handle(Message message) { |
| 209 | + // FIXME: remove me | ||
| 210 | + log.info("InternalClusterMessageHandler.handle({})", message); | ||
| 211 | + try { | ||
| 212 | + log.info("before decode"); | ||
| 204 | ClusterMessage clusterMessage = SERIALIZER.decode(message.payload()); | 213 | ClusterMessage clusterMessage = SERIALIZER.decode(message.payload()); |
| 214 | + log.info("Subject:({}), Sender:({})", clusterMessage.subject(), clusterMessage.sender()); | ||
| 205 | handler.handle(clusterMessage); | 215 | handler.handle(clusterMessage); |
| 216 | + message.respond("ACK".getBytes()); | ||
| 217 | + } catch (Exception e) { | ||
| 218 | + // TODO Auto-generated catch block | ||
| 219 | + log.error("failed", e); | ||
| 220 | + } | ||
| 206 | } | 221 | } |
| 207 | } | 222 | } |
| 208 | } | 223 | } | ... | ... |
-
Please register or login to post a comment