opendaylight/protocol_plugins/openflow_netty/src/main/java/org/opendaylight/controller/protocol_plugin/openflow/internal/FlowProgrammerService.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8
   9 package org.opendaylight.controller.protocol_plugin.openflow.internal;
  10
  11 import java.util.ArrayList;
  12 import java.util.HashMap;
  13 import java.util.HashSet;
  14 import java.util.Map;
  15 import java.util.Set;
  16 import java.util.concurrent.ConcurrentHashMap;
  17 import java.util.concurrent.ConcurrentMap;
  18
  19 import org.eclipse.osgi.framework.console.CommandInterpreter;
  20 import org.eclipse.osgi.framework.console.CommandProvider;
  21 import org.opendaylight.controller.protocol_plugin.openflow.IFlowProgrammerNotifier;
  22 import org.opendaylight.controller.protocol_plugin.openflow.IInventoryShimExternalListener;
  23 import org.opendaylight.controller.protocol_plugin.openflow.core.IController;
  24 import org.opendaylight.controller.protocol_plugin.openflow.core.IMessageListener;
  25 import org.opendaylight.controller.protocol_plugin.openflow.core.ISwitch;
  26 import org.opendaylight.controller.sal.core.ContainerFlow;
  27 import org.opendaylight.controller.sal.core.IContainerListener;
  28 import org.opendaylight.controller.sal.core.Node;
  29 import org.opendaylight.controller.sal.core.Node.NodeIDType;
  30 import org.opendaylight.controller.sal.core.NodeConnector;
  31 import org.opendaylight.controller.sal.core.Property;
  32 import org.opendaylight.controller.sal.core.UpdateType;
  33 import org.opendaylight.controller.sal.flowprogrammer.Flow;
  34 import org.opendaylight.controller.sal.flowprogrammer.IPluginInFlowProgrammerService;
  35 import org.opendaylight.controller.sal.match.Match;
  36 import org.opendaylight.controller.sal.match.MatchType;
  37 import org.opendaylight.controller.sal.utils.GlobalConstants;
  38 import org.opendaylight.controller.sal.utils.HexEncode;
  39 import org.opendaylight.controller.sal.utils.NodeCreator;
  40 import org.opendaylight.controller.sal.utils.Status;
  41 import org.opendaylight.controller.sal.utils.StatusCode;
  42 import org.openflow.protocol.OFError;
  43 import org.openflow.protocol.OFFlowMod;
  44 import org.openflow.protocol.OFFlowRemoved;
  45 import org.openflow.protocol.OFMessage;
  46 import org.openflow.protocol.OFPort;
  47 import org.openflow.protocol.OFType;
  48 import org.openflow.protocol.action.OFAction;
  49 import org.openflow.protocol.factory.MessageParseException;
  50 import org.osgi.framework.BundleContext;
  51 import org.osgi.framework.FrameworkUtil;
  52 import org.slf4j.Logger;
  53 import org.slf4j.LoggerFactory;
  54
  55 /**
  56  * Represents the openflow plugin component in charge of programming the flows
  57  * the flow programming and relay them to functional modules above SAL.
  58  */
  59 public class FlowProgrammerService implements IPluginInFlowProgrammerService,
  60         IMessageListener, IContainerListener, IInventoryShimExternalListener,
  61         CommandProvider {
  62     private static final Logger log = LoggerFactory
  63             .getLogger(FlowProgrammerService.class);
  64     private IController controller;
  65     private ConcurrentMap<String, IFlowProgrammerNotifier> flowProgrammerNotifiers;
  66     private Map<String, Set<NodeConnector>> containerToNc;
  67     private ConcurrentMap<Long, Map<Integer, Long>> xid2rid;
  68     private int barrierMessagePriorCount = getBarrierMessagePriorCount();
  69
  70     public FlowProgrammerService() {
  71         controller = null;
  72         flowProgrammerNotifiers = new ConcurrentHashMap<String, IFlowProgrammerNotifier>();
  73         containerToNc = new HashMap<String, Set<NodeConnector>>();
  74         xid2rid = new ConcurrentHashMap<Long, Map<Integer, Long>>();
  75     }
  76
  77     public void setController(IController core) {
  78         this.controller = core;
  79     }
  80
  81     public void unsetController(IController core) {
  82         if (this.controller == core) {
  83             this.controller = null;
  84         }
  85     }
  86
  87     public void setFlowProgrammerNotifier(Map<String, ?> props,
  88             IFlowProgrammerNotifier s) {
  89         if (props == null || props.get("containerName") == null) {
  90             log.error("Didn't receive the service correct properties");
  91             return;
  92         }
  93         String containerName = (String) props.get("containerName");
  94         this.flowProgrammerNotifiers.put(containerName, s);
  95     }
  96
  97     public void unsetFlowProgrammerNotifier(Map<String, ?> props,
  98             IFlowProgrammerNotifier s) {
  99         if (props == null || props.get("containerName") == null) {
 100             log.error("Didn't receive the service correct properties");
 101             return;
 102         }
 103         String containerName = (String) props.get("containerName");
 104         if (this.flowProgrammerNotifiers != null
 105                 && this.flowProgrammerNotifiers.containsKey(containerName)
 106                 && this.flowProgrammerNotifiers.get(containerName) == s) {
 107             this.flowProgrammerNotifiers.remove(containerName);
 108         }
 109     }
 110
 111     /**
 112      * Function called by the dependency manager when all the required
 113      * dependencies are satisfied
 114      *
 115      */
 116     void init() {
 117         this.controller.addMessageListener(OFType.FLOW_REMOVED, this);
 118         this.controller.addMessageListener(OFType.ERROR, this);
 119         registerWithOSGIConsole();
 120     }
 121
 122     /**
 123      * Function called by the dependency manager when at least one dependency
 124      * become unsatisfied or when the component is shutting down because for
 125      * example bundle is being stopped.
 126      *
 127      */
 128     void destroy() {
 129     }
 130
 131     /**
 132      * Function called by dependency manager after "init ()" is called and after
 133      * the services provided by the class are registered in the service registry
 134      *
 135      */
 136     void start() {
 137     }
 138
 139     /**
 140      * Function called by the dependency manager before the services exported by
 141      * the component are unregistered, this will be followed by a "destroy ()"
 142      * calls
 143      *
 144      */
 145     void stop() {
 146     }
 147
 148     @Override
 149     public Status addFlow(Node node, Flow flow) {
 150         return addFlowInternal(node, flow, 0);
 151     }
 152
 153     @Override
 154     public Status modifyFlow(Node node, Flow oldFlow, Flow newFlow) {
 155         return modifyFlowInternal(node, oldFlow, newFlow, 0);
 156     }
 157
 158     @Override
 159     public Status removeFlow(Node node, Flow flow) {
 160         return removeFlowInternal(node, flow, 0);
 161     }
 162
 163     @Override
 164     public Status addFlowAsync(Node node, Flow flow, long rid) {
 165         return addFlowInternal(node, flow, rid);
 166     }
 167
 168     @Override
 169     public Status modifyFlowAsync(Node node, Flow oldFlow, Flow newFlow,
 170             long rid) {
 171         return modifyFlowInternal(node, oldFlow, newFlow, rid);
 172     }
 173
 174     @Override
 175     public Status removeFlowAsync(Node node, Flow flow, long rid) {
 176         return removeFlowInternal(node, flow, rid);
 177     }
 178
 179     private Status addFlowInternal(Node node, Flow flow, long rid) {
 180         String action = "add";
 181         if (!node.getType().equals(NodeIDType.OPENFLOW)) {
 182             return new Status(StatusCode.NOTACCEPTABLE, errorString("send",
 183                     action, "Invalid node type"));
 184         }
 185
 186         if (controller != null) {
 187             ISwitch sw = controller.getSwitch((Long) node.getID());
 188             if (sw != null) {
 189                 FlowConverter x = new FlowConverter(flow);
 190                 OFMessage msg = x.getOFFlowMod(OFFlowMod.OFPFC_ADD, null);
 191
 192                 Object result;
 193                 if (rid == 0) {
 194                     /*
 195                      * Synchronous message send. Each message is followed by a
 196                      * Barrier message.
 197                      */
 198                     result = sw.syncSend(msg);
 199                 } else {
 200                     log.debug("Sending flow asynchronously");
 201                     /*
 202                      * Message will be sent asynchronously. A Barrier message
 203                      * will be inserted automatically to synchronize the
 204                      * progression.
 205                      */
 206                     result = asyncMsgSend(node, sw, msg, rid);
 207                 }
 208                 return getStatusInternal(result, action, rid);
 209             } else {
 210                 return new Status(StatusCode.GONE, errorString("send", action,
 211                         "Switch is not available"));
 212             }
 213         }
 214         return new Status(StatusCode.INTERNALERROR, errorString("send", action,
 215                 "Internal plugin error"));
 216     }
 217
 218     private Status modifyFlowInternal(Node node, Flow oldFlow, Flow newFlow, long rid) {
 219         String action = "modify";
 220         if (!node.getType().equals(NodeIDType.OPENFLOW)) {
 221             return new Status(StatusCode.NOTACCEPTABLE, errorString("send",
 222                     action, "Invalid node type"));
 223         }
 224         if (controller != null) {
 225             ISwitch sw = controller.getSwitch((Long) node.getID());
 226             if (sw != null) {
 227                 OFMessage msg1 = null, msg2 = null;
 228
 229                 // If priority and match portion are the same, send a
 230                 // modification message
 231                 if (oldFlow.getPriority() != newFlow.getPriority()
 232                         || !oldFlow.getMatch().equals(newFlow.getMatch())) {
 233                     msg1 = new FlowConverter(oldFlow).getOFFlowMod(
 234                             OFFlowMod.OFPFC_DELETE_STRICT, OFPort.OFPP_NONE);
 235                     msg2 = new FlowConverter(newFlow).getOFFlowMod(
 236                             OFFlowMod.OFPFC_ADD, null);
 237                 } else {
 238                     msg1 = new FlowConverter(newFlow).getOFFlowMod(
 239                             OFFlowMod.OFPFC_MODIFY_STRICT, null);
 240                 }
 241                 /*
 242                  * Synchronous message send
 243                  */
 244                 action = (msg2 == null) ? "modify" : "delete";
 245                 Object result;
 246                 if (rid == 0) {
 247                     /*
 248                      * Synchronous message send. Each message is followed by a
 249                      * Barrier message.
 250                      */
 251                     result = sw.syncSend(msg1);
 252                 } else {
 253                     /*
 254                      * Message will be sent asynchronously. A Barrier message
 255                      * will be inserted automatically to synchronize the
 256                      * progression.
 257                      */
 258                     result = asyncMsgSend(node, sw, msg1, rid);
 259                 }
 260
 261                 Status rv = getStatusInternal(result, action, rid);
 262                 if ((msg2 == null) || !rv.isSuccess()) {
 263                     return rv;
 264                 }
 265
 266                 action = "add";
 267                 if (rid == 0) {
 268                     /*
 269                      * Synchronous message send. Each message is followed by a
 270                      * Barrier message.
 271                      */
 272                     result = sw.syncSend(msg2);
 273                 } else {
 274                     /*
 275                      * Message will be sent asynchronously. A Barrier message
 276                      * will be inserted automatically to synchronize the
 277                      * progression.
 278                      */
 279                     result = asyncMsgSend(node, sw, msg2, rid);
 280                 }
 281                 return getStatusInternal(result, action, rid);
 282             } else {
 283                 return new Status(StatusCode.GONE, errorString("send", action,
 284                         "Switch is not available"));
 285             }
 286         }
 287         return new Status(StatusCode.INTERNALERROR, errorString("send", action,
 288                 "Internal plugin error"));
 289     }
 290
 291     private Status removeFlowInternal(Node node, Flow flow, long rid) {
 292         String action = "remove";
 293         if (!node.getType().equals(NodeIDType.OPENFLOW)) {
 294             return new Status(StatusCode.NOTACCEPTABLE, errorString("send",
 295                     action, "Invalid node type"));
 296         }
 297         if (controller != null) {
 298             ISwitch sw = controller.getSwitch((Long) node.getID());
 299             if (sw != null) {
 300                 OFMessage msg = new FlowConverter(flow).getOFFlowMod(
 301                         OFFlowMod.OFPFC_DELETE_STRICT, OFPort.OFPP_NONE);
 302                 Object result;
 303                 if (rid == 0) {
 304                     /*
 305                      * Synchronous message send. Each message is followed by a
 306                      * Barrier message.
 307                      */
 308                     result = sw.syncSend(msg);
 309                 } else {
 310                     /*
 311                      * Message will be sent asynchronously. A Barrier message
 312                      * will be inserted automatically to synchronize the
 313                      * progression.
 314                      */
 315                     result = asyncMsgSend(node, sw, msg, rid);
 316                 }
 317                 return getStatusInternal(result, action, rid);
 318             } else {
 319                 return new Status(StatusCode.GONE, errorString("send", action,
 320                         "Switch is not available"));
 321             }
 322         }
 323         return new Status(StatusCode.INTERNALERROR, errorString("send", action,
 324                 "Internal plugin error"));
 325     }
 326
 327     @Override
 328     public Status removeAllFlows(Node node) {
 329         return new Status(StatusCode.SUCCESS);
 330     }
 331
 332     private String errorString(String phase, String action, String cause) {
 333         return "Failed to "
 334                 + ((phase != null) ? phase + " the " + action
 335                         + " flow message: " : action + " the flow: ") + cause;
 336     }
 337
 338     @Override
 339     public void receive(ISwitch sw, OFMessage msg) {
 340         if (msg instanceof OFFlowRemoved) {
 341             handleFlowRemovedMessage(sw, (OFFlowRemoved) msg);
 342         } else if (msg instanceof OFError) {
 343             handleErrorMessage(sw, (OFError) msg);
 344         }
 345     }
 346
 347     private void handleFlowRemovedMessage(ISwitch sw, OFFlowRemoved msg) {
 348         Node node = NodeCreator.createOFNode(sw.getId());
 349         Flow flow = new FlowConverter(msg.getMatch(),
 350                 new ArrayList<OFAction>(0)).getFlow(node);
 351         flow.setPriority(msg.getPriority());
 352         flow.setIdleTimeout(msg.getIdleTimeout());
 353         flow.setId(msg.getCookie());
 354
 355         Match match = flow.getMatch();
 356         NodeConnector inPort = match.isPresent(MatchType.IN_PORT) ? (NodeConnector) match
 357                 .getField(MatchType.IN_PORT).getValue() : null;
 358
 359         for (Map.Entry<String, IFlowProgrammerNotifier> containerNotifier : flowProgrammerNotifiers
 360                 .entrySet()) {
 361             String container = containerNotifier.getKey();
 362             IFlowProgrammerNotifier notifier = containerNotifier.getValue();
 363             /*
 364              * Switch only provide us with the match information. For now let's
 365              * try to identify the container membership only from the input port
 366              * match field. In any case, upper layer consumers can derive
 367              * whether the notification was not for them. More sophisticated
 368              * filtering can be added later on.
 369              */
 370             if (inPort == null
 371                     || container.equals(GlobalConstants.DEFAULT.toString())
 372                     || this.containerToNc.get(container).contains(inPort)) {
 373                 notifier.flowRemoved(node, flow);
 374             }
 375         }
 376     }
 377
 378     private void handleErrorMessage(ISwitch sw, OFError errorMsg) {
 379         Node node = NodeCreator.createOFNode(sw.getId());
 380         OFMessage offendingMsg = null;
 381         try {
 382             offendingMsg = errorMsg.getOffendingMsg();
 383         } catch (MessageParseException e) {
 384             // TODO Auto-generated catch block
 385             e.printStackTrace();
 386         }
 387         Integer xid;
 388         if (offendingMsg != null) {
 389             xid = offendingMsg.getXid();
 390         } else {
 391             xid = errorMsg.getXid();
 392         }
 393
 394         Long rid = getMessageRid(sw.getId(), xid);
 395         /*
 396          * Null or zero requestId indicates that the error message is meant for
 397          * a sync message. It will be handled by the sync message worker thread.
 398          * Hence we are done here.
 399          */
 400         if ((rid == null) || (rid == 0)) {
 401             return;
 402         }
 403
 404         /*
 405          * Notifies the caller that error has been reported for a previous flow
 406          * programming request
 407          */
 408         for (Map.Entry<String, IFlowProgrammerNotifier> containerNotifier : flowProgrammerNotifiers
 409                 .entrySet()) {
 410             IFlowProgrammerNotifier notifier = containerNotifier.getValue();
 411             notifier.flowErrorReported(node, rid, errorMsg);
 412         }
 413     }
 414
 415     @Override
 416     public void tagUpdated(String containerName, Node n, short oldTag,
 417             short newTag, UpdateType t) {
 418
 419     }
 420
 421     @Override
 422     public void containerFlowUpdated(String containerName,
 423             ContainerFlow previousFlow, ContainerFlow currentFlow, UpdateType t) {
 424     }
 425
 426     @Override
 427     public void nodeConnectorUpdated(String containerName, NodeConnector p,
 428             UpdateType type) {
 429         Set<NodeConnector> target = null;
 430
 431         switch (type) {
 432         case ADDED:
 433             if (!containerToNc.containsKey(containerName)) {
 434                 containerToNc.put(containerName, new HashSet<NodeConnector>());
 435             }
 436             containerToNc.get(containerName).add(p);
 437             break;
 438         case CHANGED:
 439             break;
 440         case REMOVED:
 441             target = containerToNc.get(containerName);
 442             if (target != null) {
 443                 target.remove(p);
 444             }
 445             break;
 446         default:
 447         }
 448     }
 449
 450     @Override
 451     public void containerModeUpdated(UpdateType t) {
 452
 453     }
 454
 455     @Override
 456     public Status syncSendBarrierMessage(Node node) {
 457         if (!node.getType().equals(NodeIDType.OPENFLOW)) {
 458             return new Status(StatusCode.NOTACCEPTABLE,
 459                     "The node does not support Barrier message.");
 460         }
 461
 462         if (controller != null) {
 463             long swid = (Long) node.getID();
 464             ISwitch sw = controller.getSwitch(swid);
 465             if (sw != null) {
 466                 sw.syncSendBarrierMessage();
 467                 clearXid2Rid(swid);
 468                 return (new Status(StatusCode.SUCCESS));
 469             } else {
 470                 return new Status(StatusCode.GONE,
 471                         "The node does not have a valid Switch reference.");
 472             }
 473         }
 474         return new Status(StatusCode.INTERNALERROR,
 475                 "Failed to send Barrier message.");
 476     }
 477
 478     @Override
 479     public Status asyncSendBarrierMessage(Node node) {
 480         if (!node.getType().equals(NodeIDType.OPENFLOW)) {
 481             return new Status(StatusCode.NOTACCEPTABLE,
 482                     "The node does not support Barrier message.");
 483         }
 484
 485         if (controller != null) {
 486             long swid = (Long) node.getID();
 487             ISwitch sw = controller.getSwitch(swid);
 488             if (sw != null) {
 489                 sw.asyncSendBarrierMessage();
 490                 clearXid2Rid(swid);
 491                 return (new Status(StatusCode.SUCCESS));
 492             } else {
 493                 return new Status(StatusCode.GONE,
 494                         "The node does not have a valid Switch reference.");
 495             }
 496         }
 497         return new Status(StatusCode.INTERNALERROR,
 498                 "Failed to send Barrier message.");
 499     }
 500
 501     /**
 502      * This method sends the message asynchronously until the number of messages
 503      * sent reaches a threshold. Then a Barrier message is sent automatically
 504      * for sync purpose. An unique Request ID associated with the message is
 505      * passed down by the caller. The Request ID will be returned to the caller
 506      * when an error message is received from the switch.
 507      *
 508      * @param node
 509      *            The node
 510      * @param msg
 511      *            The switch
 512      * @param msg
 513      *            The OF message to be sent
 514      * @param rid
 515      *            The Request Id
 516      * @return result
 517      */
 518     private Object asyncMsgSend(Node node, ISwitch sw, OFMessage msg, long rid) {
 519         Object result = Boolean.TRUE;
 520         long swid = (Long) node.getID();
 521         int xid;
 522
 523         xid = sw.asyncSend(msg);
 524         addXid2Rid(swid, xid, rid);
 525
 526         Map<Integer, Long> swxid2rid = this.xid2rid.get(swid);
 527         if (swxid2rid == null) {
 528             return result;
 529         }
 530
 531         int size = swxid2rid.size();
 532         if (size % barrierMessagePriorCount == 0) {
 533             result = asyncSendBarrierMessage(node);
 534         }
 535
 536         return result;
 537     }
 538
 539     /**
 540      * A number of async messages are sent followed by a synchronous Barrier
 541      * message. This method returns the maximum async messages that can be sent
 542      * before the Barrier message.
 543      *
 544      * @return The max count of async messages sent prior to Barrier message
 545      */
 546     private int getBarrierMessagePriorCount() {
 547         String count = System.getProperty("of.barrierMessagePriorCount");
 548         int rv = 100;
 549
 550         if (count != null) {
 551             try {
 552                 rv = Integer.parseInt(count);
 553             } catch (Exception e) {
 554             }
 555         }
 556
 557         return rv;
 558     }
 559
 560     /**
 561      * This method returns the message Request ID previously assigned by the
 562      * caller for a given OF message xid
 563      *
 564      * @param swid
 565      *            The switch id
 566      * @param xid
 567      *            The OF message xid
 568      * @return The Request ID
 569      */
 570     private Long getMessageRid(long swid, Integer xid) {
 571         Long rid = null;
 572
 573         if (xid == null) {
 574             return rid;
 575         }
 576
 577         Map<Integer, Long> swxid2rid = this.xid2rid.get(swid);
 578         if (swxid2rid != null) {
 579             rid = swxid2rid.get(xid);
 580         }
 581         return rid;
 582     }
 583
 584     /**
 585      * This method returns a copy of outstanding xid to rid mappings.for a given
 586      * switch
 587      *
 588      * @param swid
 589      *            The switch id
 590      * @return a copy of xid2rid mappings
 591      */
 592     public Map<Integer, Long> getSwXid2Rid(long swid) {
 593         Map<Integer, Long> swxid2rid = this.xid2rid.get(swid);
 594
 595         if (swxid2rid != null) {
 596             return new HashMap<Integer, Long>(swxid2rid);
 597         } else {
 598             return new HashMap<Integer, Long>();
 599         }
 600     }
 601
 602     /**
 603      * Adds xid to rid mapping to the local DB
 604      *
 605      * @param swid
 606      *            The switch id
 607      * @param xid
 608      *            The OF message xid
 609      * @param rid
 610      *            The message Request ID
 611      */
 612     private void addXid2Rid(long swid, int xid, long rid) {
 613         Map<Integer, Long> swxid2rid = this.xid2rid.get(swid);
 614         if (swxid2rid != null) {
 615             swxid2rid.put(xid, rid);
 616         }
 617     }
 618
 619     /**
 620      * When an Error message is received, this method will be invoked to remove
 621      * the offending xid from the local DB.
 622      *
 623      * @param swid
 624      *            The switch id
 625      * @param xid
 626      *            The OF message xid
 627      */
 628     private void removeXid2Rid(long swid, int xid) {
 629         Map<Integer, Long> swxid2rid = this.xid2rid.get(swid);
 630         if (swxid2rid != null) {
 631             swxid2rid.remove(xid);
 632         }
 633     }
 634
 635     /**
 636      * Convert various result into Status
 637      *
 638      * @param result
 639      *            The returned result from previous action
 640      * @param action
 641      *            add/modify/delete flow action
 642      * @param rid
 643      *            The Request ID associated with the flow message
 644      * @return Status
 645      */
 646     private Status getStatusInternal(Object result, String action, long rid) {
 647         if (result instanceof Boolean) {
 648             return ((Boolean) result == Boolean.TRUE) ? new Status(
 649                     StatusCode.SUCCESS, rid) : new Status(
 650                     StatusCode.TIMEOUT, errorString(null, action,
 651                             "Request Timed Out"));
 652         } else if (result instanceof Status) {
 653             return (Status) result;
 654         } else if (result instanceof OFError) {
 655             OFError res = (OFError) result;
 656             return new Status(StatusCode.INTERNALERROR, errorString(
 657                     "program", action, Utils.getOFErrorString(res)));
 658         } else {
 659             return new Status(StatusCode.INTERNALERROR, errorString(
 660                     "send", action, "Internal Error"));
 661         }
 662     }
 663
 664     /**
 665      * When a Barrier reply is received, this method will be invoked to clear
 666      * the local DB
 667      *
 668      * @param swid
 669      *            The switch id
 670      */
 671     private void clearXid2Rid(long swid) {
 672         Map<Integer, Long> swxid2rid = this.xid2rid.get(swid);
 673         if (swxid2rid != null) {
 674             swxid2rid.clear();
 675         }
 676     }
 677
 678     @Override
 679     public void updateNode(Node node, UpdateType type, Set<Property> props) {
 680         long swid = (Long)node.getID();
 681
 682         switch (type) {
 683         case ADDED:
 684             Map<Integer, Long> swxid2rid = new HashMap<Integer, Long>();
 685             this.xid2rid.put(swid, swxid2rid);
 686             break;
 687         case CHANGED:
 688             break;
 689         case REMOVED:
 690             this.xid2rid.remove(swid);
 691             break;
 692         default:
 693         }
 694     }
 695
 696     @Override
 697     public void updateNodeConnector(NodeConnector nodeConnector,
 698             UpdateType type, Set<Property> props) {
 699     }
 700
 701     private void registerWithOSGIConsole() {
 702         BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass())
 703                 .getBundleContext();
 704         bundleContext.registerService(CommandProvider.class.getName(), this,
 705                 null);
 706     }
 707
 708     @Override
 709     public String getHelp() {
 710         StringBuffer help = new StringBuffer();
 711         help.append("-- Flow Programmer Service --\n");
 712         help.append("\t px2r <node id>          - Print outstanding xid2rid mappings for a given node id\n");
 713         help.append("\t px2rc                   - Print max num of async msgs prior to the Barrier\n");
 714         return help.toString();
 715     }
 716
 717     public void _px2r(CommandInterpreter ci) {
 718         String st = ci.nextArgument();
 719         if (st == null) {
 720             ci.println("Please enter a valid node id");
 721             return;
 722         }
 723
 724         long sid;
 725         try {
 726             sid = HexEncode.stringToLong(st);
 727         } catch (NumberFormatException e) {
 728             ci.println("Please enter a valid node id");
 729             return;
 730         }
 731
 732         Map<Integer, Long> swxid2rid = this.xid2rid.get(sid);
 733         if (swxid2rid == null) {
 734             ci.println("The node id entered does not exist");
 735             return;
 736         }
 737
 738         ci.println("xid             rid");
 739
 740         Set<Integer> xidSet = swxid2rid.keySet();
 741         if (xidSet == null) {
 742             return;
 743         }
 744
 745         for (Integer xid : xidSet) {
 746             ci.println(xid + "       " + swxid2rid.get(xid));
 747         }
 748     }
 749
 750     public void _px2rc(CommandInterpreter ci) {
 751         ci.println("Max num of async messages sent prior to the Barrier message is "
 752                 + barrierMessagePriorCount);
 753     }
 754 }