Send notifications using the same key every time and receive them by entryAdded() and entryUpdated()
Change-Id: I4c07bf7efc0ac7301d95140c47a15325f96dffd5
diff --git a/src/main/java/net/onrc/onos/intent/runtime/PlanInstallModule.java b/src/main/java/net/onrc/onos/intent/runtime/PlanInstallModule.java
index b8bed85..293a33d 100644
--- a/src/main/java/net/onrc/onos/intent/runtime/PlanInstallModule.java
+++ b/src/main/java/net/onrc/onos/intent/runtime/PlanInstallModule.java
@@ -43,7 +43,7 @@
private static final String PATH_INTENT_CHANNEL_NAME = "onos.pathintent";
private static final String INTENT_STATE_EVENT_CHANNEL_NAME = "onos.pathintent_state";
-
+
@Override
public void init(FloodlightModuleContext context)
throws FloodlightModuleException {
@@ -59,27 +59,27 @@
class EventListener extends Thread
implements IEventChannelListener<Long, IntentOperationList> {
-
+
private BlockingQueue<IntentOperationList> intentQueue = new LinkedBlockingQueue<>();
private Long key = Long.valueOf(0);
-
+
@Override
public void run() {
while(true) {
try {
IntentOperationList intents = intentQueue.take();
- //TODO: consider draining the remaining intent lists
+ //TODO: consider draining the remaining intent lists
// and processing in one big batch
// List<IntentOperationList> remaining = new LinkedList<>();
// intentQueue.drainTo(remaining);
-
+
processIntents(intents);
} catch (InterruptedException e) {
log.warn("Error taking from intent queue: {}", e.getMessage());
}
}
}
-
+
private void processIntents(IntentOperationList intents) {
log("start_processIntents");
log.debug("Processing OperationList {}", intents);
@@ -90,13 +90,13 @@
log("begin_installPlan");
boolean success = planInstall.installPlan(plan);
log("end_installPlan");
-
+
log("begin_sendInstallNotif");
sendNotifications(intents, true, success);
log("end_sendInstallNotif");
log("finish");
}
-
+
private void sendNotifications(IntentOperationList intents, boolean installed, boolean success) {
IntentStateList states = new IntentStateList();
for(IntentOperation i : intents) {
@@ -123,11 +123,23 @@
states.put(i.intent.getId(), newState);
}
intentStateChannel.addEntry(key, states);
- key += 1;
+ // XXX: Send notifications using the same key every time
+ // and receive them by entryAdded() and entryUpdated()
+ // key += 1;
}
-
+
@Override
public void entryAdded(IntentOperationList value) {
+ entryUpdated(value);
+ }
+
+ @Override
+ public void entryRemoved(IntentOperationList value) {
+ // This channel is a queue, so this method is not needed
+ }
+
+ @Override
+ public void entryUpdated(IntentOperationList value) {
log("start_intentNotifRecv");
log("begin_sendReceivedNotif");
sendNotifications(value, false, false);
@@ -141,36 +153,26 @@
log.warn("Error putting to intent queue: {}", e.getMessage());
}
}
-
- @Override
- public void entryRemoved(IntentOperationList value) {
- // This channel is a queue, so this method is not needed
- }
-
- @Override
- public void entryUpdated(IntentOperationList value) {
- // This channel is a queue, so this method is not needed
- }
}
-
+
public static void log(String step) {
log.error("Time:{}, Step:{}", System.nanoTime(), step);
}
-
+
@Override
public void startUp(FloodlightModuleContext context) {
// start subscriber
- datagridService.addListener(PATH_INTENT_CHANNEL_NAME,
- eventListener,
- Long.class,
+ datagridService.addListener(PATH_INTENT_CHANNEL_NAME,
+ eventListener,
+ Long.class,
IntentOperationList.class);
eventListener.start();
// start publisher
- intentStateChannel = datagridService.createChannel(INTENT_STATE_EVENT_CHANNEL_NAME,
- Long.class,
+ intentStateChannel = datagridService.createChannel(INTENT_STATE_EVENT_CHANNEL_NAME,
+ Long.class,
IntentStateList.class);
}
-
+
@Override
public Collection<Class<? extends IFloodlightService>> getModuleDependencies() {
Collection<Class<? extends IFloodlightService>> l =
@@ -181,7 +183,7 @@
l.add(IFlowPusherService.class);
return l;
}
-
+
@Override
public Collection<Class<? extends IFloodlightService>> getModuleServices() {
// no services, for now