Refactor _callbacks and _push_watchers init to Connection

cassandra/connection.py

@@ -194,6 +194,7 @@ class Connection(object):
         self.is_control_connection = is_control_connection
         self.user_type_map = user_type_map
         self._push_watchers = defaultdict(set)
+        self._callbacks = {}
         self._iobuf = io.BytesIO()
         if protocol_version >= 3:
             self._header_unpack = v3_header_unpack

cassandra/io/asyncorereactor.py

@@ -158,7 +158,6 @@ class AsyncoreConnection(Connection, asyncore.dispatcher):
         Connection.__init__(self, *args, **kwargs)
         asyncore.dispatcher.__init__(self)
 
-        self._callbacks = {}
         self.deque = deque()
         self.deque_lock = Lock()
 

cassandra/io/eventletreactor.py

@@ -85,9 +85,6 @@ class EventletConnection(Connection):
 
         self._write_queue = Queue()
 
-        self._callbacks = {}
-        self._push_watchers = defaultdict(set)
-
         sockerr = None
         addresses = socket.getaddrinfo(
             self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM

cassandra/io/geventreactor.py

@@ -82,9 +82,6 @@ class GeventConnection(Connection):
 
         self._write_queue = Queue()
 
-        self._callbacks = {}
-        self._push_watchers = defaultdict(set)
-
         sockerr = None
         addresses = socket.getaddrinfo(self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM)
         for (af, socktype, proto, canonname, sockaddr) in addresses:

cassandra/io/libevreactor.py

@@ -264,7 +264,6 @@ class LibevConnection(Connection):
     def __init__(self, *args, **kwargs):
         Connection.__init__(self, *args, **kwargs)
 
-        self._callbacks = {}
         self.deque = deque()
         self._deque_lock = Lock()
 

cassandra/io/twistedreactor.py

@@ -189,7 +189,6 @@ class TwistedConnection(Connection):
         self.is_closed = True
         self.connector = None
 
-        self._callbacks = {}
         reactor.callFromThread(self.add_connection)
         self._loop.maybe_start()
 

tests/unit/test_connection.py

@@ -261,10 +261,7 @@ class ConnectionTest(unittest.TestCase):
         Ensure the following methods throw NIE's. If not, come back and test them.
         """
         c = self.make_connection()
-
         self.assertRaises(NotImplementedError, c.close)
-        self.assertRaises(NotImplementedError, c.register_watcher, None, None)
-        self.assertRaises(NotImplementedError, c.register_watchers, None)
 
     def test_set_keyspace_blocking(self):
         c = self.make_connection()