MOTOR-938 Docs for watch() incorrectly call ChangeStream.close() (#163)

doc/examples/tornado_change_stream_example.py

@@ -68,7 +68,10 @@ def on_change(cls, change):
 
         # Each change notification has a binary _id. Use it to make an HTML
         # element id, then remove it.
-        html_id = urlsafe_b64encode(change["_id"]["_data"]).decode().rstrip("=")
+        data = change["_id"]["_data"]
+        if not isinstance(data, bytes):
+            data = data.encode("utf-8")
+        html_id = urlsafe_b64encode(data).decode().rstrip("=")
         change.pop("_id")
         change["html"] = '<div id="change-%s"><pre>%s</pre></div>' % (
             html_id,
@@ -104,15 +107,13 @@ def main():
     app = Application()
     app.listen(options.port)
     loop = tornado.ioloop.IOLoop.current()
+
     # Start watching collection for changes.
-    loop.add_callback(watch, collection)
     try:
-        loop.start()
+        loop.run_sync(lambda: watch(collection))
     except KeyboardInterrupt:
-        pass
-    finally:
-        if change_stream is not None:
-            change_stream.close()
+        if change_stream:
+            loop.run_sync(change_stream.close)
 
 
 if __name__ == "__main__":

motor/core.py

@@ -1065,23 +1065,17 @@ async def watch_collection():
           def main():
               loop = IOLoop.current()
               # Start watching collection for changes.
-              loop.add_callback(watch_collection)
-              try:
-                  loop.start()
-              except KeyboardInterrupt:
-                  pass
-              finally:
-                  if change_stream is not None:
-                      change_stream.close()
+            try:
+                loop.run_sync(watch_collection)
+            except KeyboardInterrupt:
+                if change_stream:
+                   loop.run_sync(change_stream.close)
 
           # asyncio
           try:
-              asyncio.run(watch_collection)
+              asyncio.run(watch_collection())
           except KeyboardInterrupt:
-                  pass
-          finally:
-              if change_stream is not None:
-                  change_stream.close()
+              pass
 
         The :class:`~MotorChangeStream` async iterable blocks
         until the next change document is returned or an error is raised. If