diff --git a/src/osmo_gsm_tester/obj/bsc_osmo.py b/src/osmo_gsm_tester/obj/bsc_osmo.py
index 158bf93..510063a 100644
--- a/src/osmo_gsm_tester/obj/bsc_osmo.py
+++ b/src/osmo_gsm_tester/obj/bsc_osmo.py
@@ -149,8 +149,10 @@
         # over this list, we have a 1:1 match in indexes.
         return self.bts.index(bts)
 
-    def bts_is_connected(self, bts):
-        return OsmoBscCtrl(self).bts_is_connected(self.bts_num(bts))
+    def bts_is_connected(self, bts, use_ctrl=None):
+        if use_ctrl is None:
+            use_ctrl = self.ctrl()
+        return use_ctrl.bts_is_connected(self.bts_num(bts))
 
     def running(self):
         return not self.process.terminated()
@@ -160,32 +162,16 @@
             self.vty.disconnect()
             self.vty = None
 
-class OsmoBscCtrl(log.Origin):
-    PORT = 4249
-    BTS_OML_STATE_VAR = "bts.%d.oml-connection-state"
-    BTS_OML_STATE_RE = re.compile("GET_REPLY (\d+) bts.\d+.oml-connection-state (?P<oml_state>\w+)")
-
-    def __init__(self, bsc):
-        self.bsc = bsc
-        super().__init__(log.C_BUS, 'CTRL(%s:%d)' % (self.bsc.addr(), OsmoBscCtrl.PORT))
-
     def ctrl(self):
-        return osmo_ctrl.OsmoCtrl(self.bsc.addr(), OsmoBscCtrl.PORT)
+        return OsmoBscCtrl(self)
+
+class OsmoBscCtrl(osmo_ctrl.OsmoCtrl):
+    def __init__(self, bsc, port=4249):
+        self.bsc = bsc
+        super().__init__(bsc.addr(), port)
 
     def bts_is_connected(self, bts_num):
-        with self.ctrl() as ctrl:
-            ctrl.do_get(OsmoBscCtrl.BTS_OML_STATE_VAR % bts_num)
-            data = ctrl.receive()
-            while (len(data) > 0):
-                (answer, data) = ctrl.remove_ipa_ctrl_header(data)
-                answer_str = answer.decode('utf-8')
-                answer_str = answer_str.replace('\n', ' ')
-                res = OsmoBscCtrl.BTS_OML_STATE_RE.match(answer_str)
-                if res:
-                    oml_state = str(res.group('oml_state'))
-                    if oml_state == 'connected':
-                        return True
-        return False
+        return self.get_var('bts.%d.oml-connection-state' % bts_num) == 'connected'
 
 class OsmoBscVty(osmo_vty.OsmoVty):
     def __init__(self, bsc, port=4242):
diff --git a/src/osmo_gsm_tester/obj/msc_osmo.py b/src/osmo_gsm_tester/obj/msc_osmo.py
index 67e1d31..5a7c0ba 100644
--- a/src/osmo_gsm_tester/obj/msc_osmo.py
+++ b/src/osmo_gsm_tester/obj/msc_osmo.py
@@ -157,29 +157,12 @@
         return not self.process.terminated()
 
 
-class OsmoMscCtrl(log.Origin):
-    PORT = 4255
-    SUBSCR_LIST_ACTIVE_VAR = 'subscriber-list-active-v1'
-
-    def __init__(self, msc):
+class OsmoMscCtrl(osmo_ctrl.OsmoCtrl):
+    def __init__(self, msc, port=4255):
         self.msc = msc
-        super().__init__(log.C_BUS, 'CTRL(%s:%d)' % (self.msc.addr(), self.PORT))
-
-    def ctrl(self):
-        return osmo_ctrl.OsmoCtrl(self.msc.addr(), self.PORT)
+        super().__init__(self.msc.addr(), port)
 
     def subscriber_list_active(self):
-        aslist_str = ""
-        with self.ctrl() as ctrl:
-            ctrl.do_get(self.SUBSCR_LIST_ACTIVE_VAR)
-            # This is legacy code from the old osmo-gsm-tester.
-            # looks like this doesn't work for long data.
-            data = ctrl.receive()
-            while (len(data) > 0):
-                (answer, data) = ctrl.remove_ipa_ctrl_header(data)
-                answer_str = answer.decode('utf-8')
-                answer_str = answer_str.replace('\n', ' ')
-                aslist_str = answer_str
-            return aslist_str
+        return self.get_var('subscriber-list-active-v1').replace('\n', ' ')
 
 # vim: expandtab tabstop=4 shiftwidth=4
diff --git a/src/osmo_gsm_tester/obj/nitb_osmo.py b/src/osmo_gsm_tester/obj/nitb_osmo.py
index a424927..ea00a75 100644
--- a/src/osmo_gsm_tester/obj/nitb_osmo.py
+++ b/src/osmo_gsm_tester/obj/nitb_osmo.py
@@ -158,22 +158,10 @@
         return not self.process.terminated()
 
 
-class OsmoNitbCtrl(log.Origin):
-    PORT = 4249
-    SUBSCR_MODIFY_VAR = 'subscriber-modify-v1'
-    SUBSCR_MODIFY_REPLY_RE = re.compile("SET_REPLY (\d+) %s OK" % SUBSCR_MODIFY_VAR)
-    SUBSCR_DELETE_VAR = 'subscriber-delete-v1'
-    SUBSCR_DELETE_REPLY_RE = re.compile("SET_REPLY (\d+) %s Removed" % SUBSCR_DELETE_VAR)
-    SUBSCR_LIST_ACTIVE_VAR = 'subscriber-list-active-v1'
-    BTS_OML_STATE_VAR = "bts.%d.oml-connection-state"
-    BTS_OML_STATE_RE = re.compile("GET_REPLY (\d+) bts.\d+.oml-connection-state (?P<oml_state>\w+)")
-
-    def __init__(self, nitb):
+class OsmoNitbCtrl(osmo_ctrl.OsmoCtrl):
+    def __init__(self, nitb, port=4249):
         self.nitb = nitb
-        super().__init__(log.C_BUS, 'CTRL(%s:%d)' % (self.nitb.addr(), OsmoNitbCtrl.PORT))
-
-    def ctrl(self):
-        return osmo_ctrl.OsmoCtrl(self.nitb.addr(), OsmoNitbCtrl.PORT)
+        super().__init__(nitb.addr(), port)
 
     def subscriber_add(self, imsi, msisdn, ki=None, algo=None):
         if algo:
@@ -181,54 +169,17 @@
         else:
             value = '%s,%s' % (imsi, msisdn)
 
-        with self.ctrl() as ctrl:
-            ctrl.do_set(OsmoNitbCtrl.SUBSCR_MODIFY_VAR, value)
-            data = ctrl.receive()
-            (answer, data) = ctrl.remove_ipa_ctrl_header(data)
-            answer_str = answer.decode('utf-8')
-            res = OsmoNitbCtrl.SUBSCR_MODIFY_REPLY_RE.match(answer_str)
-            if not res:
-                raise RuntimeError('Cannot create subscriber %r (answer=%r)' % (imsi, answer_str))
-            self.dbg('Created subscriber', imsi=imsi, msisdn=msisdn)
+        assert self.set_var('subscriber-modify-v1', value) == 'OK'
+        self.dbg('Created subscriber', imsi=imsi, msisdn=msisdn)
 
     def subscriber_delete(self, imsi):
-        with self.ctrl() as ctrl:
-            ctrl.do_set(OsmoNitbCtrl.SUBSCR_DELETE_VAR, imsi)
-            data = ctrl.receive()
-            (answer, data) = ctrl.remove_ipa_ctrl_header(data)
-            answer_str = answer.decode('utf-8')
-            res = OsmoNitbCtrl.SUBSCR_DELETE_REPLY_RE.match(answer_str)
-            if not res:
-                raise RuntimeError('Cannot delete subscriber %r (answer=%r)' % (imsi, answer_str))
-            self.dbg('Deleted subscriber', imsi=imsi)
+        assert self.set_var('subscriber-delete-v1', imsi) == 'Removed'
+        self.dbg('Deleted subscriber', imsi=imsi)
 
     def subscriber_list_active(self):
-        aslist_str = ""
-        with self.ctrl() as ctrl:
-            ctrl.do_get(OsmoNitbCtrl.SUBSCR_LIST_ACTIVE_VAR)
-            # This is legacy code from the old osmo-gsm-tester.
-            # looks like this doesn't work for long data.
-            data = ctrl.receive()
-            while (len(data) > 0):
-                (answer, data) = ctrl.remove_ipa_ctrl_header(data)
-                answer_str = answer.decode('utf-8')
-                answer_str = answer_str.replace('\n', ' ')
-                aslist_str = answer_str
-            return aslist_str
+        return self.get_var('subscriber-list-active-v1').replace('\n', ' ')
 
     def bts_is_connected(self, bts_num):
-        with self.ctrl() as ctrl:
-            ctrl.do_get(OsmoNitbCtrl.BTS_OML_STATE_VAR % bts_num)
-            data = ctrl.receive()
-            while (len(data) > 0):
-                (answer, data) = ctrl.remove_ipa_ctrl_header(data)
-                answer_str = answer.decode('utf-8')
-                answer_str = answer_str.replace('\n', ' ')
-                res = OsmoNitbCtrl.BTS_OML_STATE_RE.match(answer_str)
-                if res:
-                    oml_state = str(res.group('oml_state'))
-                    if oml_state == 'connected':
-                        return True
-        return False
+        return self.get_var('bts.%d.oml-connection-state' % bts_num) == 'connected'
 
 # vim: expandtab tabstop=4 shiftwidth=4
diff --git a/src/osmo_gsm_tester/obj/osmo_ctrl.py b/src/osmo_gsm_tester/obj/osmo_ctrl.py
index c2dd7e3..644025f 100644
--- a/src/osmo_gsm_tester/obj/osmo_ctrl.py
+++ b/src/osmo_gsm_tester/obj/osmo_ctrl.py
@@ -20,8 +20,20 @@
 
 import socket
 import struct
+import re
 
 from ..core import log
+from ..core.event_loop import MainLoop
+
+VERB_SET = 'SET'
+VERB_GET = 'GET'
+VERB_SET_REPLY = 'SET_REPLY'
+VERB_GET_REPLY = 'GET_REPLY'
+VERB_TRAP = 'TRAP'
+VERB_ERROR = 'ERROR'
+RECV_VERBS = (VERB_GET_REPLY, VERB_SET_REPLY, VERB_TRAP, VERB_ERROR)
+recv_re = re.compile('(%s) ([0-9]+) (.*)' % ('|'.join(RECV_VERBS)),
+                     re.MULTILINE + re.DOTALL)
 
 class CtrlInterfaceExn(Exception):
     pass
@@ -56,41 +68,168 @@
             raise CtrlInterfaceExn("Wrong protocol in answer!")
         return data[4:plen+3], data[plen+3:]
 
-    def connect(self):
-        self.dbg('Connecting')
-        self.sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.sck.connect((self.host, self.port))
+    def try_connect(self):
+        '''Do a connection attempt, return True when successful, False otherwise.
+           Does not raise exceptions, but logs them to the debug log.'''
+        assert self.sck is None
+        try:
+            self.dbg('Connecting')
+            sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            try:
+                sck.connect((self.host, self.port))
+            except:
+                sck.close()
+                raise
+            # set self.sck only after the connect was successful
+            self.sck = sck
+            return True
+        except:
+            self.dbg('Failed to connect', sys.exc_info()[0])
+            return False
+
+    def connect(self, timeout=30):
+        '''Connect to the CTRL self.host and self.port, retry for 'timeout' seconds.'''
+        MainLoop.wait(self.try_connect, timestep=3, timeout=timeout)
         self.sck.setblocking(1)
         self.sck.settimeout(10)
 
     def disconnect(self):
+        if self.sck is None:
+            return
         self.dbg('Disconnecting')
-        if self.sck is not None:
-            self.sck.close()
+        self.sck.close()
+        self.sck = None
 
-    def _send(self, data):
+    def _recv(self, verbs, match_args=None, match_id=None, attempts=10, length=1024):
+        '''Receive until a response matching the verbs / args / msg-id is obtained from CTRL.
+           The general socket timeout applies for each attempt made, see connect().
+           Multiple attempts may be necessary if, for example, intermediate
+           messages are received that do not relate to what is expected, like
+           TRAPs that are not interesting.
+
+           To receive a GET_REPLY / SET_REPLY:
+             verb, rx_id, val = _recv(('GET_REPLY', 'ERROR'), match_id=used_id)
+             if verb == 'ERROR':
+                 raise CtrlInterfaceExn()
+             print(val)
+
+           To receive a TRAP:
+             verb, rx_id, val = _recv('TRAP', 'bts_connection_status connected')
+             # val == 'bts_connection_status connected'
+
+           If the CTRL is not connected yet, open and close a connection for
+           this operation only.
+        '''
+
+        # allow calling for both already connected VTY as well as establishing
+        # a connection just for this command.
+        if self.sck is None:
+            with self:
+                return self._recv(verbs, match_args=match_args,
+                        match_id=match_id, attempts=attempts, length=length)
+
+        if isinstance(verbs, str):
+            verbs = (verbs, )
+
+        for i in range(attempts):
+            data = self.sck.recv(length)
+            self.dbg('Receiving', data=data)
+            while len(data) > 0:
+                msg, data = self.remove_ipa_ctrl_header(data)
+                msg_str = msg.decode('utf-8')
+
+                m = recv_re.fullmatch(msg_str)
+                if m is None:
+                    raise CtrlInterfaceExn('Received garbage: %r' % data)
+
+                rx_verb, rx_id, rx_args = m.groups()
+                rx_id = int(rx_id)
+
+                if match_id is not None and match_id != rx_id:
+                    continue
+
+                if verbs and rx_verb not in verbs:
+                    continue
+
+                if match_args and not rx_args.startswith(match_args):
+                    continue
+
+                return rx_verb, rx_id, rx_args
+        raise CtrlInterfaceExn('No answer found: ' + reply_header)
+
+    def _sendrecv(self, verb, send_args, *recv_args, use_id=None, **recv_kwargs):
+        '''Send a request and receive a matching response.
+           If the CTRL is not connected yet, open and close a connection for
+           this operation only.
+        '''
+        if self.sck is None:
+            with self:
+                return self._sendrecv(verb, send_args, *recv_args, use_id=use_id, **recv_kwargs)
+
+        if use_id is None:
+            use_id = self.next_id()
+
+        # send
+        data = '{verb} {use_id} {send_args}'.format(**locals())
         self.dbg('Sending', data=data)
         data = self.prefix_ipa_ctrl_header(data)
         self.sck.send(data)
 
-    def receive(self, length = 1024):
-        data = self.sck.recv(length)
-        self.dbg('Receiving', data=data)
-        return data
+        # receive reply
+        recv_kwargs['match_id'] = use_id
+        return self._recv(*recv_args, **recv_kwargs)
 
-    def do_set(self, var, value, use_id=None):
-        if use_id is None:
-            use_id = self.next_id()
-        setmsg = "SET %s %s %s" %(use_id, var, value)
-        self._send(setmsg)
-        return use_id
+    def set_var(self, var, value):
+        '''Set the value of a specific variable on a CTRL interface, and return the response, e.g.:
+              assert set_var('subscriber-modify-v1', '901701234567,2342') == 'OK'
+           If the CTRL is not connected yet, open and close a connection for
+           this operation only.
+        '''
+        verb, rx_id, args = self._sendrecv(VERB_SET, '%s %s' % (var, value), (VERB_SET_REPLY, VERB_ERROR))
 
-    def do_get(self, var, use_id=None):
-        if use_id is None:
-            use_id = self.next_id()
-        getmsg = "GET %s %s" %(use_id, var)
-        self._send(getmsg)
-        return use_id
+        if verb == VERB_ERROR:
+            raise CtrlInterfaceExn('SET %s = %s returned %r' % (var, value, ' '.join((verb, str(rx_id), args))))
+
+        var_and_space = var + ' '
+        if not args.startswith(var_and_space):
+            raise CtrlInterfaceExn('SET %s = %s returned SET_REPLY for different var: %r'
+                                   % (var, value, ' '.join((verb, str(rx_id), args))))
+
+        return args[len(var_and_space):]
+
+    def get_var(self, var):
+        '''Get the value of a specific variable from a CTRL interface:
+              assert get_var('bts.0.oml-connection-state') == 'connected'
+           If the CTRL is not connected yet, open and close a connection for
+           this operation only.
+        '''
+        verb, rx_id, args = self._sendrecv(VERB_GET, var, (VERB_GET_REPLY, VERB_ERROR))
+
+        if verb == VERB_ERROR:
+            raise CtrlInterfaceExn('GET %s returned %r' % (var, ' '.join((verb, str(rx_id), args))))
+
+        var_and_space = var + ' '
+        if not args.startswith(var_and_space):
+            raise CtrlInterfaceExn('GET %s returned GET_REPLY for different var: %r'
+                                   % (var, value, ' '.join((verb, str(rx_id), args))))
+
+        return args[len(var_and_space):]
+
+    def get_int_var(self, var):
+        '''Same as get_var() but return an int'''
+        return int(self.get_var(var))
+
+    def get_trap(self, name):
+        '''Read from CTRL until a TRAP of this name is received.
+           If name is None, any TRAP is returned.
+           If the CTRL is not connected yet, open and close a connection for
+           this operation only.
+        '''
+        verb, rx_id, args = self._recv(VERB_TRAP, name)
+        name_and_space = var + ' '
+        # _recv() should ensure this:
+        assert args.startswith(name_and_space)
+        return args[len(name_and_space):]
 
     def __enter__(self):
         self.connect()