ryu.app.ofctl: implement multipart

an example:
    msg = parser.OFPPortDescStatsRequest(datapath=datapath)
    result = ofctl.api.send_msg(self, msg,
                                reply_cls=parser.OFPPortDescStatsReply,
                                reply_multi=True)

Signed-off-by: YAMAMOTO Takashi <yamamoto@valinux.co.jp>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

ryu/app/ofctl/api.py

@@ -29,12 +29,13 @@ def get_datapath(app, dpid):
     return app.send_request(event.GetDatapathRequest(dpid=dpid))()
 
 
-def send_msg(app, msg, reply_cls=None):
+def send_msg(app, msg, reply_cls=None, reply_multi=False):
     """
     Send an openflow message.
     """
     return app.send_request(event.SendMsgRequest(msg=msg,
-                                                 reply_cls=reply_cls))()
+                                                 reply_cls=reply_cls,
+                                                 reply_multi=reply_multi))()
 
 
 app_manager.require_app('ryu.app.ofctl.service')

ryu/app/ofctl/event.py

@@ -40,17 +40,21 @@ class GetDatapathRequest(_RequestBase):
 # send msg
 
 class SendMsgRequest(_RequestBase):
-    def __init__(self, msg, reply_cls=None):
+    def __init__(self, msg, reply_cls=None, reply_multi=False):
         super(SendMsgRequest, self).__init__()
         self.msg = msg
         self.reply_cls = reply_cls
+        self.reply_multi = reply_multi
 
 
 # generic reply
 
 class Reply(_ReplyBase):
-    def __init__(self, result=None):
+    def __init__(self, result=None, exception=None):
         self.result = result
+        self.exception = exception
 
     def __call__(self):
+        if self.exception:
+            raise self.exception
         return self.result

ryu/app/ofctl/exception.py

@@ -0,0 +1,29 @@
+# Copyright (C) 2014 Nippon Telegraph and Telephone Corporation.
+# Copyright (C) 2014 YAMAMOTO Takashi <yamamoto at valinux co jp>
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ryu import exception
+
+
+# base classes
+
+class _ExceptionBase(exception.RyuException):
+    def __init__(self, result):
+        self.result = result
+        super(_ExceptionBase, self).__init__(result=result)
+
+
+class UnexpectedMultiReply(_ExceptionBase):
+    message = 'Unexpected Multi replies %(result)s'

ryu/app/ofctl/service.py

@@ -24,6 +24,7 @@ from ryu.controller.handler import CONFIG_DISPATCHER, MAIN_DISPATCHER,\
 from ryu.controller.handler import set_ev_cls
 
 import event
+import exception
 
 
 class _SwitchInfo(object):
@@ -103,19 +104,26 @@ class OfctlService(app_manager.RyuApp):
     def _handle_send_msg(self, req):
         if not req.reply_cls is None:
             self._observe_msg(req.reply_cls)
+
         msg = req.msg
         datapath = msg.datapath
         datapath.set_xid(msg)
         xid = msg.xid
-        datapath.send_msg(msg)
         barrier = datapath.ofproto_parser.OFPBarrierRequest(datapath)
         datapath.set_xid(barrier)
         barrier_xid = barrier.xid
-        datapath.send_msg(barrier)
+
         si = self._switches[datapath.id]
+        assert not xid in si.results
+        assert not xid in si.xids
+        assert not barrier_xid in si.barriers
+        si.results[xid] = []
         si.xids[xid] = req
         si.barriers[barrier_xid] = xid
 
+        datapath.send_msg(msg)
+        datapath.send_msg(barrier)
+
     @set_ev_cls(ofp_event.EventOFPBarrierReply, MAIN_DISPATCHER)
     def _handle_barrier(self, ev):
         msg = ev.msg
@@ -126,18 +134,23 @@ class OfctlService(app_manager.RyuApp):
             self.logger.error('unknown dpid %s' % (datapath.id,))
             return
         try:
-            xid = si.barriers[msg.xid]
+            xid = si.barriers.pop(msg.xid)
         except KeyError:
             self.logger.error('unknown barrier xid %s' % (msg.xid,))
             return
-        try:
-            result = si.results.pop(xid)
-        except KeyError:
-            result = None
+        result = si.results.pop(xid)
         req = si.xids.pop(xid)
         if not req.reply_cls is None:
             self._unobserve_msg(req.reply_cls)
-        rep = event.Reply(result=result)
+        if req.reply_multi:
+            rep = event.Reply(result=result)
+        elif len(result) == 0:
+            rep = event.Reply()
+        elif len(result) == 1:
+            rep = event.Reply(result=result[0])
+        else:
+            rep = event.Reply(exception=exception.
+                              UnexpectedMultiReply(result=result))
         self.reply_to_request(req, rep)
 
     @set_ev_cls(ofp_event.EventOFPErrorMsg, MAIN_DISPATCHER)
@@ -160,6 +173,6 @@ class OfctlService(app_manager.RyuApp):
                               (ev, msg.xid,))
             return
         try:
-            si.results[msg.xid] = ev.msg
+            si.results[msg.xid].append(ev.msg)
         except KeyError:
             self.logger.error('unknown error xid %s' % (msg.xid,))