ws_topology: bugfix an event is dropped

When SocketError occurs, an event may not be notified to an other client.
It is because rpc_clients is changed in the loop.

Signed-off-by: Satoshi Kobayashi <satoshi-k@stratosphere.co.jp>
Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

ryu/app/ws_topology.py

@@ -35,7 +35,8 @@ $ sudo mn --controller=remote --topo linear,2
 """  # noqa
 
 from socket import error as SocketError
-from tinyrpc.exc import InvalidReplyError
+from ryu.contrib.tinyrpc.exc import InvalidReplyError
+
 
 from ryu.app.wsgi import (
     ControllerBase,
@@ -83,6 +84,7 @@ class WebSocketTopology(app_manager.RyuApp):
         self._rpc_broadcall('event_link_delete', msg)
 
     def _rpc_broadcall(self, func_name, msg):
+        disconnected_clients = []
         for rpc_client in self.rpc_clients:
             # NOTE: Although broadcasting is desired,
             #       RPCClient#get_proxy(one_way=True) does not work well
@@ -91,10 +93,13 @@ class WebSocketTopology(app_manager.RyuApp):
                 getattr(rpc_server, func_name)(msg)
             except SocketError:
                 self.logger.debug('WebSocket disconnected: %s' % rpc_client.ws)
-                self.rpc_clients.remove(rpc_client)
+                disconnected_clients.append(rpc_client)
             except InvalidReplyError as e:
                 self.logger.error(e)
 
+        for client in disconnected_clients:
+            self.rpc_clients.remove(client)
+
 
 class WebSocketTopologyController(ControllerBase):
 

ryu/tests/unit/app/test_ws_topology.py

@@ -0,0 +1,54 @@
+# Copyright (C) 2013 Stratosphere Inc.
+#
+# 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.
+
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+import unittest
+from socket import error as SocketError
+
+import mock
+
+from ryu.app.ws_topology import WebSocketTopology
+
+
+class Test_ws_topology(unittest.TestCase):
+
+    def test_when_sock_error(self):
+        args = {
+            'wsgi': mock.Mock(),
+        }
+        app = WebSocketTopology(**args)
+
+        rpc_client_mock1 = mock.Mock()
+        config = {
+            'get_proxy.return_value.event_link_add.side_effect': SocketError,
+        }
+        rpc_client_mock1.configure_mock(**config)
+
+        rpc_client_mock2 = mock.Mock()
+
+        app.rpc_clients = [
+            rpc_client_mock1,
+            rpc_client_mock2,
+        ]
+
+        ev_mock = mock.Mock()
+        app._event_link_add_handler(ev_mock)
+
+        rpc_client_mock1.get_proxy.assert_called_once_with()
+        rpc_client_mock2.get_proxy.assert_called_once_with()
+
+if __name__ == "__main__":
+    unittest.main()