Repository: bluesky/bluesky (python)

Issue:
no stop document ^C after  a previous ^C and typing `RE.stop()` or `RE.abort()`
<!--- Provide a general summary of the issue in the Title above -->
If `^C` is issued by the user after `RE.stop()` or `RE.abort()`, bluesky does not respond.  This may happen because the user made a mistake and should have typed `RE.hault()` or simply thought they could stop every thing from moving for convenience (probably a habit of users of another common DAQ software).

Whether the users should or should not do this for convenience is probably not a topic for discussion here.  What is important to note is that pressing `^C` many times seems to break the `RE` and the terminal freezes.  Once this happens, one needs to exit the session in order to kill the process, and as a consequence, the stop document is never written to the database.

## Expected Behavior
stop document gets written to the database after pressing `Enter` for `RE.stop()` ,  `RE.abort()` or `RE.hault()`?  I don't see anything else in the stop document that happens after this, but an expert should confirm this (or others with more advanced usage that is beyond out-of-the-box bluesky)

## Current Behavior
stop document gets written to the database after as soon as possible (not waiting on motors to return to the position dictated by underlying plan code for exceptions since this doesn't seem to be a critical part of the stop document

## Possible Solution
see expected behavior, but any solution anyone one dreams of is ok by me.

## Steps to Reproduce (for bugs)
see above

## Context
see above

## Your Environment
`{'bluesky': '1.8.3.post1+g34931bb9', 'ophyd': '1.6.4'}` on RHEL8 


Pull request:
Fix/msg mutator
<!--- Provide a general summary of your changes in the Title above -->

## Description
<!--- Describe your changes in detail -->

Correctly handle the exception communication channel in `msg_mutator`.

Fixes code from 3dd2ef03d73fc9b614abff1724a55a06ba74374b (2016). 

## Motivation and Context
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here. -->

closes #1626 and #1625 

## How Has This Been Tested?
<!--- Please describe in detail how you tested your changes. -->
<!--- Include details of your testing environment, and the tests you ran to -->
<!--- see how your change affects other areas of the code, etc. -->

Two new tests added:

 - one "pure" test that checks mechanical changes to a sequence of messages
 - one "integration" test that verifies that for `rel_scan` the close run message (which creates the stop document) will come out before the motors are returned to their initial position.

There are incidental changes make pytest not fail on me for pytest reasons locally.

<!--
## Screenshots (if appropriate):
-->


Patch:
diff --git a/bluesky/preprocessors.py b/bluesky/preprocessors.py
index 859bdcfd..5c92cb54 100644
--- a/bluesky/preprocessors.py
+++ b/bluesky/preprocessors.py
@@ -214,7 +214,7 @@ def plan_mutator(plan, msg_proc):
 
 def msg_mutator(plan, msg_proc):
     """
-    A simple preprocessor that mutates or deletes single messages in a plan
+    A simple preprocessor that mutates or deletes single messages in a plan.
 
     To *insert* messages, use ``plan_mutator`` instead.
 
@@ -234,20 +234,38 @@ def msg_mutator(plan, msg_proc):
     --------
     :func:`bluesky.plans.plan_mutator`
     """
-    ret = None
-    while True:
-        try:
-            msg = plan.send(ret)
-            msg = msg_proc(msg)
-            # if None, just skip message
-            # feed 'None' back down into the base plan,
-            # this may break some plans
-            if msg is None:
-                ret = None
-                continue
-            ret = yield msg
-        except StopIteration as e:
-            return e.value
+    try:
+        msg = plan.send(None)
+    except StopIteration as _e:
+        ret = _e.value
+    else:
+        while 1:
+            try:
+                msg = msg_proc(msg)
+                # if None, just skip message
+                # feed 'None' back down into the base plan,
+                # this may break some plans
+                if msg is None:
+                    _s = None
+                else:
+                    _s = yield msg
+            except GeneratorExit:
+                plan.close()
+                raise
+            except BaseException as _e:
+                try:
+                    msg = plan.throw(_e)
+                except StopIteration as _e:
+                    ret = _e.value
+                    break
+            else:
+                try:
+                    msg = plan.send(_s)
+                except StopIteration as _e:
+                    ret = _e.value
+                    break
+
+    return ret
 
 
 def pchain(*args):
@@ -289,7 +307,10 @@ def print_summary_wrapper(plan):
     """
 
     read_cache = []
-    for msg in plan:
+
+    def spy(msg):
+        nonlocal read_cache
+
         cmd = msg.command
         if cmd == 'open_run':
             print('{:=^80}'.format(' Open Run '))
@@ -304,7 +325,9 @@ def print_summary_wrapper(plan):
             read_cache.append(msg.obj.name)
         elif cmd == 'save':
             print('  Read {}'.format(read_cache))
-        yield msg
+        return msg
+
+    return (yield from msg_mutator(plan, spy))
 
 
 def run_wrapper(plan, *, md=None):


Test patch:
diff --git a/bluesky/tests/test_bec.py b/bluesky/tests/test_bec.py
index 49a818a1..71193fb3 100644
--- a/bluesky/tests/test_bec.py
+++ b/bluesky/tests/test_bec.py
@@ -1,3 +1,5 @@
+import warnings
+
 import ast
 import pytest
 import jsonschema
@@ -214,10 +216,10 @@ def test_plot_ints(RE):
     RE(bps.mov(s, int(0)))
     assert s.describe()['s']['dtype'] == 'integer'
     s.kind = 'hinted'
-    with pytest.warns(None) as record:
-        RE(count([s], num=35))
 
-    assert len(record) == 0
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        RE(count([s], num=35))
 
 
 def test_plot_prune_fifo(RE, hw):
@@ -326,9 +328,9 @@ def test_many_motors(RE, hw):
     bec = BestEffortCallback()
     RE.subscribe(bec)
     movement = [(motor, 1, 5, 5) for motor in motors]
-    with pytest.warns(None) as record:
+    with pytest.warns((RuntimeWarning, UserWarning)):
         RE(grid_scan(dets, *[item for sublist in movement for item in sublist]))
-    assert len(record) > 0
+
     assert not bec._live_plots
     assert not bec._live_grids
     assert not bec._live_scatters
diff --git a/bluesky/tests/test_plans.py b/bluesky/tests/test_plans.py
index a969c0ea..207e4865 100644
--- a/bluesky/tests/test_plans.py
+++ b/bluesky/tests/test_plans.py
@@ -5,6 +5,7 @@ from bluesky.tests.utils import DocCollector
 import bluesky.plans as bp
 import bluesky.plan_stubs as bps
 import bluesky.preprocessors as bpp
+from bluesky.utils import RequestStop
 import numpy as np
 import numpy.testing as npt
 import pandas as pd
@@ -714,3 +715,37 @@ def test_describe_failure(RE):
             st
         )
     st.verify()
+
+
+def test_errors_through_msg_mutator(hw):
+    gen = bp.rel_scan([], hw.motor, 5, -5, 10)
+
+    msgs = []
+
+    msgs.append(next(gen))
+    msgs.append(next(gen))
+    msgs.append(next(gen))
+    msgs.append(next(gen))
+    msgs.append(next(gen))
+    msgs.append(gen.throw(RequestStop))
+    try:
+        while True:
+            msgs.append(next(gen))
+    except RequestStop:
+        pass
+    else:
+        raise False
+
+    target = [
+        "stage",
+        "open_run",
+        "declare_stream",
+        "checkpoint",
+        "set",
+        "close_run",
+        "unstage",
+        "set",
+        "wait",
+    ]
+
+    assert target == [m.command for m in msgs]
diff --git a/bluesky/tests/test_preprocessors.py b/bluesky/tests/test_preprocessors.py
index 9b1279b5..e29d5fb9 100644
--- a/bluesky/tests/test_preprocessors.py
+++ b/bluesky/tests/test_preprocessors.py
@@ -1,6 +1,10 @@
 import bluesky.plan_stubs as bps
-from bluesky.run_engine import RunEngine
-from bluesky.preprocessors import contingency_decorator
+from bluesky.run_engine import RunEngine, RequestStop
+from bluesky.preprocessors import (
+    contingency_decorator,
+    msg_mutator,
+    contingency_wrapper,
+)
 import pytest
 from unittest.mock import MagicMock
 
@@ -77,3 +81,38 @@ def test_given_a_plan_that_raises_contigency_with_no_auto_raise_and_except_plan_
         assert exception == expected_exception
 
     except_plan.assert_called_once()
+
+
+def test_exceptions_through_msg_mutator():
+    from bluesky import Msg
+
+    def outer():
+        for j in range(50):
+            yield Msg(f"step {j}")
+
+    def attach(msg):
+        cmd = msg.command
+        return msg._replace(command=f"{cmd}+")
+
+    def except_plan(e):
+        yield Msg("handle it")
+
+    gen = msg_mutator(contingency_wrapper(outer(), except_plan=except_plan), attach)
+
+    msgs = []
+
+    msgs.append(next(gen))
+    msgs.append(next(gen))
+    msgs.append(next(gen))
+    msgs.append(next(gen))
+    msgs.append(gen.throw(RequestStop))
+    try:
+        while True:
+            msgs.append(next(gen))
+    except RequestStop:
+        pass
+    else:
+        raise False
+    assert ["step 0+", "step 1+", "step 2+", "step 3+", "handle it+"] == [
+        m.command for m in msgs
+    ]