diff --git a/src/debugpy/__init__.py b/src/debugpy/__init__.py
index 5a4cca43..dfd809ba 100644
--- a/src/debugpy/__init__.py
+++ b/src/debugpy/__init__.py
@@ -15,6 +15,7 @@ __all__ = [
     "__version__",
     "breakpoint",
     "configure",
+    "connect",
     "debug_this_thread",
     "is_client_connected",
     "listen",
diff --git a/src/debugpy/adapter/__main__.py b/src/debugpy/adapter/__main__.py
index e01bfec9..503805d4 100644
--- a/src/debugpy/adapter/__main__.py
+++ b/src/debugpy/adapter/__main__.py
@@ -49,24 +49,24 @@ def main(args):
     if args.for_server is None:
         adapter.access_token = compat.force_str(codecs.encode(os.urandom(32), "hex"))
 
+    endpoints = {}
     try:
-        server_host, server_port = servers.serve()
+        client_host, client_port = clients.serve(args.host, args.port)
     except Exception as exc:
         if args.for_server is None:
             raise
-        endpoints = {"error": "Can't listen for server connections: " + str(exc)}
+        endpoints = {"error": "Can't listen for client connections: " + str(exc)}
     else:
-        endpoints = {"server": {"host": server_host, "port": server_port}}
-        try:
-            client_host, client_port = clients.serve(args.host, args.port)
-        except Exception as exc:
-            if args.for_server is None:
-                raise
-            endpoints = {"error": "Can't listen for client connections: " + str(exc)}
-        else:
-            endpoints["client"] = {"host": client_host, "port": client_port}
+        endpoints["client"] = {"host": client_host, "port": client_port}
 
     if args.for_server is not None:
+        try:
+            server_host, server_port = servers.serve()
+        except Exception as exc:
+            endpoints = {"error": "Can't listen for server connections: " + str(exc)}
+        else:
+            endpoints["server"] = {"host": server_host, "port": server_port}
+
         log.info(
             "Sending endpoints info to debug server at localhost:{0}:\n{1!j}",
             args.for_server,
@@ -101,7 +101,9 @@ def main(args):
             try:
                 os.remove(listener_file)
             except Exception:
-                log.swallow_exception("Failed to delete {0!r}", listener_file, level="warning")
+                log.swallow_exception(
+                    "Failed to delete {0!r}", listener_file, level="warning"
+                )
 
         try:
             with open(listener_file, "w") as f:
@@ -149,6 +151,10 @@ def _parse_argv(argv):
         help="start the adapter in debugServer mode on the specified host",
     )
 
+    parser.add_argument(
+        "--access-token", type=str, help="access token expected from the server"
+    )
+
     parser.add_argument(
         "--server-access-token", type=str, help="access token expected by the server"
     )
diff --git a/src/debugpy/adapter/clients.py b/src/debugpy/adapter/clients.py
index 2a285308..39ae6b65 100644
--- a/src/debugpy/adapter/clients.py
+++ b/src/debugpy/adapter/clients.py
@@ -8,6 +8,7 @@ import os
 import sys
 
 import debugpy
+from debugpy import adapter
 from debugpy.common import fmt, json, log, messaging, sockets
 from debugpy.common.compat import unicode
 from debugpy.adapter import components, servers, sessions
@@ -275,15 +276,23 @@ class Client(components.Component):
         )
         console_title = request("consoleTitle", json.default("Python Debug Console"))
 
-        launchers.spawn_debuggee(
-            self.session, request, args, console, console_title
-        )
+        servers.serve()
+        launchers.spawn_debuggee(self.session, request, args, console, console_title)
 
     @_start_message_handler
     def attach_request(self, request):
         if self.session.no_debug:
             raise request.isnt_valid('"noDebug" is not supported for "attach"')
 
+        listen = request("listen", False)
+        if listen:
+            host = request("host", "127.0.0.1")
+            port = request("port", int)
+            adapter.access_token = None
+            host, port = servers.serve(host, port)
+        else:
+            host, port = servers.serve()
+
         # There are four distinct possibilities here.
         #
         # If "processId" is specified, this is attach-by-PID. We need to inject the
@@ -294,12 +303,12 @@ class Client(components.Component):
         # in response to a "debugpyAttach" event. If so, the debug server should be
         # connected already, and thus the wait timeout is zero.
         #
-        # If neither is specified, and "waitForAttach" is true, this is attach-by-socket
+        # If neither is specified, and "listen" is true, this is attach-by-socket
         # with the server expected to connect to the adapter via debugpy.connect(). There
         # is no PID known in advance, so just wait until the first server connection
         # indefinitely, with no timeout.
         #
-        # If neither is specified, and "waitForAttach" is false, this is attach-by-socket
+        # If neither is specified, and "listen" is false, this is attach-by-socket
         # in which the server has spawned the adapter via debugpy.listen(). There
         # is no PID known to the client in advance, but the server connection should be
         # either be there already, or the server should be connecting shortly, so there
@@ -330,11 +339,12 @@ class Client(components.Component):
         else:
             if sub_pid == ():
                 pred = lambda conn: True
-                timeout = None if request("waitForAttach", False) else 10
+                timeout = None if listen else 10
             else:
                 pred = lambda conn: conn.pid == sub_pid
                 timeout = 0
 
+        self.channel.send_event("debugpyWaitingForServer", {"host": host, "port": port})
         conn = servers.wait_for_connection(self.session, pred, timeout)
         if conn is None:
             raise request.cant_handle(
@@ -436,15 +446,17 @@ class Client(components.Component):
             body = dict(self.start_request.arguments)
             self._known_subprocesses.add(conn)
 
+        body.pop("processId", None)
+        if body.pop("listen", False):
+            body.pop("host", None)
+            body.pop("port", None)
         body["name"] = fmt("Subprocess {0}", conn.pid)
         body["request"] = "attach"
+        body["subProcessId"] = conn.pid
         if "host" not in body:
             body["host"] = "127.0.0.1"
         if "port" not in body:
             _, body["port"] = listener.getsockname()
-        if "processId" in body:
-            del body["processId"]
-        body["subProcessId"] = conn.pid
 
         self.channel.send_event("debugpyAttach", body)
 
diff --git a/src/debugpy/adapter/servers.py b/src/debugpy/adapter/servers.py
index c7e54aeb..ab7554a0 100644
--- a/src/debugpy/adapter/servers.py
+++ b/src/debugpy/adapter/servers.py
@@ -344,9 +344,9 @@ class Server(components.Component):
         super(Server, self).disconnect()
 
 
-def serve():
+def serve(host="127.0.0.1", port=0):
     global listener
-    listener = sockets.serve("Server", Connection, "127.0.0.1")
+    listener = sockets.serve("Server", Connection, host, port)
     return listener.getsockname()
 
 
diff --git a/src/debugpy/server/cli.py b/src/debugpy/server/cli.py
index a99195cb..311eb959 100644
--- a/src/debugpy/server/cli.py
+++ b/src/debugpy/server/cli.py
@@ -26,7 +26,7 @@ TARGET = "<filename> | -m <module> | -c <code> | --pid <pid>"
 HELP = """debugpy {0}
 See https://aka.ms/debugpy for documentation.
 
-Usage: debugpy --listen [<address>:]<port>
+Usage: debugpy [--listen | --connect] [<address>:]<port>
                [--wait-for-client]
                [--configure-<name> <value>]...
                [--log-to <path>] [--log-to-stderr]
diff --git a/tests/debug/config.py b/tests/debug/config.py
index eff43b7a..d42e7c81 100644
--- a/tests/debug/config.py
+++ b/tests/debug/config.py
@@ -65,6 +65,7 @@ class DebugConfig(collections.MutableMapping):
         # Attach by socket
         "host": (),
         "port": (),
+        "listen": False,
         # Attach by PID
         "processId": (),
     }
diff --git a/tests/debug/runners.py b/tests/debug/runners.py
index 3b00a3fc..c4776e2b 100644
--- a/tests/debug/runners.py
+++ b/tests/debug/runners.py
@@ -19,10 +19,10 @@ be bound to specific arguments, by using either [] or with_options(), which can
 chained arbitrarily::
 
     # Direct invocation:
-    session.attach_by_socket("cli", log_dir="...")
+    session.attach_connect("cli", log_dir="...")
 
     # Indirect invocation:
-    run = runners.attach_by_socket
+    run = runners.attach_connect
     run = run["cli"]
     run = run.with_options(log_dir="...")
     run(session, target)
@@ -59,12 +59,13 @@ import sys
 
 import debugpy
 from debugpy.common import compat, fmt, log
-from tests import net
+from tests import net, timeline
 from tests.debug import session
+from tests.patterns import some
 
 
 def _runner(f):
-    assert f.__name__.startswith("launch") or f.__name__.startswith("attach")
+    # assert f.__name__.startswith("launch") or f.__name__.startswith("attach")
     setattr(session.Session, f.__name__, f)
 
     class Runner(object):
@@ -159,7 +160,7 @@ def _attach_common_config(session, target, cwd):
 
 @_runner
 @contextlib.contextmanager
-def attach_by_pid(session, target, cwd=None, wait=True):
+def attach_pid(session, target, cwd=None, wait=True):
     if sys.version_info < (3,) and sys.platform == "darwin":
         pytest.skip("https://github.com/microsoft/ptvsd/issues/1916")
     if wait and not sys.platform.startswith("linux"):
@@ -188,7 +189,7 @@ while "debugpy" not in sys.modules:
 
 from debuggee import scratchpad
 
-while "_attach_by_pid" not in scratchpad:
+while "_attach_pid" not in scratchpad:
     time.sleep(0.1)
     """
         else:
@@ -202,24 +203,20 @@ while "_attach_by_pid" not in scratchpad:
         yield
 
     if wait:
-        session.scratchpad["_attach_by_pid"] = True
+        session.scratchpad["_attach_pid"] = True
 
 
 @_runner
-def attach_by_socket(
-    session, target, method, listener="server", cwd=None, wait=True, log_dir=None
-):
+def attach_listen(session, target, method, cwd=None, wait=True, log_dir=None):
     log.info(
         "Attaching {0} to {1} by socket using {2}.", session, target, method.upper()
     )
 
     assert method in ("api", "cli")
-    assert listener in ("server")  # TODO: ("adapter", "server")
 
     config = _attach_common_config(session, target, cwd)
-
-    host = config["host"] = attach_by_socket.host
-    port = config["port"] = attach_by_socket.port
+    config["host"] = host = attach_listen.host
+    config["port"] = port = attach_listen.port
 
     if method == "cli":
         args = [
@@ -257,8 +254,57 @@ if {wait!r}:
     return session.request_attach()
 
 
-attach_by_socket.host = "127.0.0.1"
-attach_by_socket.port = net.get_test_server_port(5678, 5800)
+attach_listen.host = "127.0.0.1"
+attach_listen.port = net.get_test_server_port(5678, 5800)
+
+
+@_runner
+def attach_connect(session, target, method, cwd=None, log_dir=None):
+    log.info(
+        "Attaching {0} to {1} by socket using {2}.", session, target, method.upper()
+    )
+
+    assert method in ("api", "cli")
+
+    config = _attach_common_config(session, target, cwd)
+    config["listen"] = True
+    config["host"] = host = attach_connect.host
+    config["port"] = port = attach_connect.port
+
+    if method == "cli":
+        args = [
+            os.path.dirname(debugpy.__file__),
+            "--connect",
+            compat.filename_str(host) + ":" + str(port),
+        ]
+        if log_dir is not None:
+            args += ["--log-to", log_dir]
+        debuggee_setup = None
+    elif method == "api":
+        args = []
+        debuggee_setup = """
+import debugpy
+if {log_dir!r}:
+    debugpy.log_to({log_dir!r})
+debugpy.connect({address!r})
+"""
+        debuggee_setup = fmt(debuggee_setup, address=(host, port), log_dir=log_dir)
+    else:
+        raise ValueError
+    args += target.cli(session.spawn_debuggee.env)
+
+    def spawn_debuggee(occ):
+        assert occ.body == some.dict.containing({"host": host, "port": port})
+        session.spawn_debuggee(args, cwd=cwd, setup=debuggee_setup)
+
+    session.timeline.when(timeline.Event("debugpyWaitingForServer"), spawn_debuggee)
+    session.spawn_adapter(args=[] if log_dir is None else ["--log-dir", log_dir])
+    return session.request_attach()
+
+
+attach_connect.host = "127.0.0.1"
+attach_connect.port = net.get_test_server_port(5678, 5800)
+
 
 all_launch = [
     launch["internalConsole"],
@@ -266,8 +312,18 @@ all_launch = [
     launch["externalTerminal"],
 ]
 
-all_attach_by_socket = [attach_by_socket["api"], attach_by_socket["cli"]]
+all_attach_listen = [
+    attach_listen["api"],
+    attach_listen["cli"],
+]
 
-all_attach = all_attach_by_socket + [attach_by_pid]
+all_attach_connect = [
+    attach_connect["api"],
+    attach_connect["cli"],
+]
+
+all_attach_socket = all_attach_listen + all_attach_connect
+
+all_attach = all_attach_socket + [attach_pid]
 
 all = all_launch + all_attach
diff --git a/tests/debug/session.py b/tests/debug/session.py
index 556c2421..d07a075a 100644
--- a/tests/debug/session.py
+++ b/tests/debug/session.py
@@ -166,8 +166,7 @@ class Session(object):
             [
                 timeline.Event("module"),
                 timeline.Event("continued"),
-                # timeline.Event("exited"),
-                # timeline.Event("terminated"),
+                timeline.Event("debugpyWaitingForServer"),
                 timeline.Event("thread", some.dict.containing({"reason": "started"})),
                 timeline.Event("thread", some.dict.containing({"reason": "exited"})),
                 timeline.Event("output", some.dict.containing({"category": "stdout"})),
@@ -390,11 +389,11 @@ class Session(object):
         while not self.adapter_endpoints.check():
             time.sleep(0.1)
 
-    def spawn_adapter(self):
+    def spawn_adapter(self, args=()):
         assert self.adapter is None
         assert self.channel is None
 
-        args = [sys.executable, os.path.dirname(debugpy.adapter.__file__)]
+        args = [sys.executable, os.path.dirname(debugpy.adapter.__file__)] + list(args)
         env = self._make_env(self.spawn_adapter.env)
 
         log.info(
diff --git a/tests/debugpy/test_attach.py b/tests/debugpy/test_attach.py
index 230d497d..e89c3979 100644
--- a/tests/debugpy/test_attach.py
+++ b/tests/debugpy/test_attach.py
@@ -53,7 +53,7 @@ def test_attach_api(pyfile, target, wait_for_client, is_client_connected, stop_m
                 time.sleep(0.1)
 
     with debug.Session() as session:
-        host, port = runners.attach_by_socket.host, runners.attach_by_socket.port
+        host, port = runners.attach_connect.host, runners.attach_connect.port
         session.config.update({"host": host, "port": port})
 
         backchannel = session.open_backchannel()
@@ -97,7 +97,7 @@ def test_attach_api(pyfile, target, wait_for_client, is_client_connected, stop_m
         session.request_continue()
 
 
-@pytest.mark.parametrize("run", runners.all_attach_by_socket)
+@pytest.mark.parametrize("run", runners.all_attach_listen)
 def test_reattach(pyfile, target, run):
     @pyfile
     def code_to_debug():
@@ -146,7 +146,7 @@ def test_reattach(pyfile, target, run):
 
 
 @pytest.mark.parametrize("pid_type", ["int", "str"])
-def test_attach_by_pid_client(pyfile, target, pid_type):
+def test_attach_pid_client(pyfile, target, pid_type):
     @pyfile
     def code_to_debug():
         import debuggee
@@ -178,7 +178,7 @@ def test_attach_by_pid_client(pyfile, target, pid_type):
     session1.captured_output = set()
     session1.expected_exit_code = None  # not expected to exit on disconnect
 
-    with session1.attach_by_pid(target(code_to_debug), wait=False):
+    with session1.attach_pid(target(code_to_debug), wait=False):
         session1.set_breakpoints(code_to_debug, all)
 
     session1.wait_for_stop(expected_frames=[some.dap.frame(code_to_debug, "bp")])
@@ -191,7 +191,7 @@ def test_attach_by_pid_client(pyfile, target, pid_type):
     session1.wait_for_terminated()
 
     with debug.Session() as session2:
-        with session2.attach_by_pid(pid, wait=False):
+        with session2.attach_pid(pid, wait=False):
             session2.set_breakpoints(code_to_debug, all)
 
         stop = session2.wait_for_stop(
diff --git a/tests/debugpy/test_breakpoints.py b/tests/debugpy/test_breakpoints.py
index d31df207..bc0352b8 100644
--- a/tests/debugpy/test_breakpoints.py
+++ b/tests/debugpy/test_breakpoints.py
@@ -17,7 +17,7 @@ from tests.patterns import some
 bp_root = test_data / "bp"
 
 
-@pytest.fixture(params=[runners.launch, runners.attach_by_socket["api"]])
+@pytest.fixture(params=[runners.launch, runners.attach_listen["api"]])
 def run(request):
     return request.param
 
diff --git a/tests/debugpy/test_disconnect.py b/tests/debugpy/test_disconnect.py
index 07e6cce0..cb0425d1 100644
--- a/tests/debugpy/test_disconnect.py
+++ b/tests/debugpy/test_disconnect.py
@@ -12,7 +12,7 @@ from tests.debug import runners
 from tests.patterns import some
 
 
-@pytest.mark.parametrize("run", runners.all_attach_by_socket)
+@pytest.mark.parametrize("run", runners.all_attach_socket)
 def test_continue_on_disconnect_for_attach(pyfile, target, run):
     @pyfile
     def code_to_debug():
diff --git a/tests/debugpy/test_django.py b/tests/debugpy/test_django.py
index 2378e721..bc00864b 100644
--- a/tests/debugpy/test_django.py
+++ b/tests/debugpy/test_django.py
@@ -28,7 +28,7 @@ class lines:
 
 
 @pytest.fixture
-@pytest.mark.parametrize("run", [runners.launch, runners.attach_by_socket["cli"]])
+@pytest.mark.parametrize("run", [runners.launch, runners.attach_listen["cli"]])
 def start_django(run):
     def start(session, multiprocess=False):
         # No clean way to kill Django server, expect non-zero exit code
diff --git a/tests/debugpy/test_flask.py b/tests/debugpy/test_flask.py
index 5cdbb0ce..b3b9282b 100644
--- a/tests/debugpy/test_flask.py
+++ b/tests/debugpy/test_flask.py
@@ -30,7 +30,7 @@ class lines:
 
 
 @pytest.fixture
-@pytest.mark.parametrize("run", [runners.launch, runners.attach_by_socket["cli"]])
+@pytest.mark.parametrize("run", [runners.launch, runners.attach_listen["cli"]])
 def start_flask(run):
     def start(session, multiprocess=False):
         # No clean way to kill Flask server, expect non-zero exit code
diff --git a/tests/debugpy/test_gevent.py b/tests/debugpy/test_gevent.py
index f9afe2bc..4026354c 100644
--- a/tests/debugpy/test_gevent.py
+++ b/tests/debugpy/test_gevent.py
@@ -60,7 +60,7 @@ def test_gevent(pyfile, target, run):
     with debug.Session() as session:
         session.config["gevent"] = True
 
-        if str(run) == "attach_by_socket(cli)":
+        if str(run) == "listen(cli)" or str(run) == "connect(cli)":
             session.spawn_debuggee.env["GEVENT_SUPPORT"] = "True"
 
         with run(session, target(code_to_debug)):
diff --git a/tests/debugpy/test_log.py b/tests/debugpy/test_log.py
index 1259f3c4..6fcf1ee3 100644
--- a/tests/debugpy/test_log.py
+++ b/tests/debugpy/test_log.py
@@ -13,9 +13,9 @@ from tests.debug import runners, targets
 
 @contextlib.contextmanager
 def check_logs(tmpdir, run, pydevd_log):
-    # For attach_by_pid, there's ptvsd.server process that performs the injection,
+    # For attach_pid, there's ptvsd.server process that performs the injection,
     # and then there's the debug server that is injected into the debuggee.
-    server_count = 2 if type(run).__name__ == "attach_by_pid" else 1
+    server_count = 2 if type(run).__name__ == "attach_pid" else 1
 
     expected_logs = {
         "debugpy.adapter-*.log": 1,
@@ -33,18 +33,18 @@ def check_logs(tmpdir, run, pydevd_log):
     assert actual_logs() == expected_logs
 
 
+@pytest.mark.parametrize("run", runners.all_attach_socket)
 @pytest.mark.parametrize("target", targets.all)
-@pytest.mark.parametrize("method", ["api", "cli"])
-def test_log_dir(pyfile, tmpdir, target, method):
+def test_log_dir(pyfile, tmpdir, run, target):
     @pyfile
     def code_to_debug():
         import debuggee
 
         debuggee.setup()
 
-    # Depending on the method, attach_by_socket will use either `debugpy --log-dir ...`
+    # Depending on the method, the runner will use either `debugpy --log-dir ...`
     # or `debugpy.log_to() ...`.
-    run = runners.attach_by_socket[method].with_options(log_dir=tmpdir.strpath)
+    run = run.with_options(log_dir=tmpdir.strpath)
     with check_logs(tmpdir, run, pydevd_log=False):
         with debug.Session() as session:
             session.log_dir = None
diff --git a/tests/debugpy/test_multiproc.py b/tests/debugpy/test_multiproc.py
index 8718c42a..53e66573 100644
--- a/tests/debugpy/test_multiproc.py
+++ b/tests/debugpy/test_multiproc.py
@@ -13,7 +13,7 @@ from tests.debug import runners
 from tests.patterns import some
 
 
-@pytest.fixture(params=[runners.launch, runners.attach_by_socket["api"]])
+@pytest.fixture(params=[runners.launch] + runners.all_attach_socket)
 def run(request):
     return request.param
 
@@ -101,6 +101,7 @@ def test_multiprocessing(pyfile, target, run, start_method):
             pass
 
         expected_child_config = dict(parent_session.config)
+        expected_child_config.pop("listen", None)
         expected_child_config.update(
             {
                 "name": some.str,
@@ -120,6 +121,7 @@ def test_multiprocessing(pyfile, target, run, start_method):
                 pass
 
             expected_grandchild_config = dict(child_session.config)
+            expected_grandchild_config.pop("listen", None)
             expected_grandchild_config.update(
                 {
                     "name": some.str,
@@ -181,6 +183,7 @@ def test_subprocess(pyfile, target, run):
             pass
 
         expected_child_config = dict(parent_session.config)
+        expected_child_config.pop("listen", None)
         expected_child_config.update(
             {
                 "name": some.str,
diff --git a/tests/debugpy/test_source_mapping.py b/tests/debugpy/test_source_mapping.py
index b1b2c958..b7f2feb4 100644
--- a/tests/debugpy/test_source_mapping.py
+++ b/tests/debugpy/test_source_mapping.py
@@ -12,7 +12,7 @@ from tests.debug import runners
 from tests.patterns import some
 
 
-@pytest.fixture(params=[runners.launch, runners.attach_by_socket["api"]])
+@pytest.fixture(params=[runners.launch, runners.attach_listen["api"]])
 def run(request):
     return request.param
 
diff --git a/tests/pytest_fixtures.py b/tests/pytest_fixtures.py
index fe9e2bc2..db3f2f89 100644
--- a/tests/pytest_fixtures.py
+++ b/tests/pytest_fixtures.py
@@ -20,7 +20,7 @@ from tests.debug import runners, session, targets
 
 if int(os.environ.get("DEBUGPY_TESTS_FULL", "0")):
     TARGETS = targets.all_named
-    RUNNERS = runners.all_launch + runners.all_attach_by_socket
+    RUNNERS = runners.all_launch + runners.all_attach_socket
 else:
     TARGETS = [targets.Program]
     RUNNERS = [runners.launch]
diff --git a/tests/timeline.py b/tests/timeline.py
index 5440c602..9d06cb23 100644
--- a/tests/timeline.py
+++ b/tests/timeline.py
@@ -30,6 +30,7 @@ class Timeline(object):
         self.name = str(name if name is not None else id(self))
         self.ignore_unobserved = []
 
+        self._listeners = []  # [(expectation, callable)]
         self._index_iter = itertools.count(1)
         self._accepting_new = threading.Event()
         self._finalized = threading.Event()
@@ -341,6 +342,10 @@ class Timeline(object):
                 self._recorded_new.notify_all()
                 self._record_queue.task_done()
 
+                for exp, callback in tuple(self._listeners):
+                    if exp == occ:
+                        callback(occ)
+
     def mark(self, id, block=True):
         occ = Occurrence("mark", id)
         occ.id = id
@@ -360,6 +365,12 @@ class Timeline(object):
         occ = ResponseOccurrence(request_occ, message)
         return self._record(occ, block)
 
+    def when(self, expectation, callback):
+        """For every occurrence recorded after this call, invokes callback(occurrence)
+        if occurrence == expectation.
+        """
+        self._listeners.append((expectation, callback))
+
     def _snapshot(self):
         last = self._last
         occ = self._beginning