Merge branch 'master' into issue-697

.github/workflows/python-package.yml

@@ -33,7 +33,7 @@ jobs:
     strategy:
       matrix:
         platform: [ubuntu-latest, macos-latest, windows-latest]
-        python-version: [3.7, 3.8, 3.9, "3.10", pypy-3.8]
+        python-version: [3.7, 3.8, 3.9, "3.10", 3.11, pypy-3.8]
     runs-on: ${{ matrix.platform }}
 
     steps:

.pre-commit-config.yaml

@@ -26,7 +26,7 @@ repos:
         language: node
         pass_filenames: false
         types: [python]
-        additional_dependencies: ["[email protected].254"]
+        additional_dependencies: ["[email protected].286"]
     repo: local
   - hooks:
       - id: mypy

docs/operators.rst

@@ -26,16 +26,18 @@ Operator                                                   Description
 Transforming Observables
 ------------------------
 
-================================================   ================================================
-Operator                                                           Description
-================================================   ================================================
-:func:`buffer <reactivex.operators.buffer>`        Periodically gather items from an Observable into bundles and emit these bundles rather than emitting the items one at a time.
-:func:`flat_map <reactivex.operators.flat_map>`    Transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable.
-:func:`group_by <reactivex.operators.group_by>`    Divide an Observable into a set of Observables that each emit a different group of items from the original Observable, organized by key.
-:func:`map <reactivex.operators.map>`              Transform the items emitted by an Observable by applying a function to each item.
-:func:`scan <reactivex.operators.scan>`            Apply a function to each item emitted by an Observable, sequentially, and emit each successive value.
-:func:`window <reactivex.operators.window>`        Periodically subdivide items from an Observable into Observable windows and emit these windows rather than emitting the items one at a time.
-================================================   ================================================
+================================================    ================================================
+Operator                                                            Description
+================================================    ================================================
+:func:`buffer <reactivex.operators.buffer>`         Periodically gather items from an Observable into bundles and emit these bundles rather than emitting the items one at a time.
+:func:`flat_map <reactivex.operators.flat_map>`     Transform the items emitted by an Observable into Observables, then flatten the emissions from those into a single Observable.
+:func:`concat_map <reactivex.operators.concat_map>` Projects each source value to an Observable which is merged in the output Observable, in a serialized fashion waiting for each one to complete before merging the next.
+:func:`switch_map <reactivex.operators.switch_map>` Projects each source value to an Observable which is merged in the output Observable, emitting values only from the most recently projected Observable.
+:func:`group_by <reactivex.operators.group_by>`     Divide an Observable into a set of Observables that each emit a different group of items from the original Observable, organized by key.
+:func:`map <reactivex.operators.map>`               Transform the items emitted by an Observable by applying a function to each item.
+:func:`scan <reactivex.operators.scan>`             Apply a function to each item emitted by an Observable, sequentially, and emit each successive value.
+:func:`window <reactivex.operators.window>`         Periodically subdivide items from an Observable into Observable windows and emit these windows rather than emitting the items one at a time.
+================================================    ================================================
 
 Filtering Observables
 ----------------------

docs/testing.rst

@@ -16,18 +16,13 @@ Basic example
     from reactivex.testing import ReactiveTest, TestScheduler
     from reactivex import operators
 
-    # setting up aliases for more concise code
-    on_next = ReactiveTest.on_next
-    on_error = ReactiveTest.on_error
-    on_completed = ReactiveTest.on_completed
-
     def test_double():
         # Create a scheduler
         scheduler = TestScheduler()
         # Define one or more source
         source = scheduler.create_hot_observable(
-            on_next(250, 3),
-            on_next(350, 5),
+            ReactiveTest.on_next(250, 3),
+            ReactiveTest.on_next(350, 5),
         )
 
         # Define how the observable/operator is used on the source
@@ -39,8 +34,8 @@ Basic example
 
         # check the messages and potentially subscriptions
         assert results.messages == [
-            on_next(250, 6),
-            on_next(350, 10),
+            ReactiveTest.on_next(250, 6),
+            ReactiveTest.on_next(350, 10),
         ]
 
 
@@ -53,6 +48,10 @@ or with full control, you can easily test various situations and combinations
 .. _in_sequence_or_throw:
 
 .. code:: python
+    # setting up aliases for more concise code
+    on_next = ReactiveTest.on_next
+    on_error = ReactiveTest.on_error
+    on_completed = ReactiveTest.on_completed
 
     def test_operator():
         # Code to test; takes a sequence of integers and passes through,
@@ -72,9 +71,8 @@ or with full control, you can easily test various situations and combinations
         source = scheduler.create_cold_observable(
             on_next(300, 1), on_next(400, 2), on_next(500, 3), on_completed(600)
         )
-        # Here is another way to create the same observable, 
-        # as long as we set the correct scheduler
-        source = reactivex.from_marbles('------1-2-3-|', timespan=50, scheduler=scheduler)
+        # Here is another way to create the same observable
+        source = reactivex.from_marbles('------1-2-3-|', timespan=50)
         # You can shorten the "create" function from the basic example to a lambda with no arguments
         result = scheduler.start(lambda: source.pipe(
             in_sequence_or_throw(),
@@ -91,20 +89,20 @@ Timeline
 
 When ``scheduler.start`` is called, the test scheduler starts moving its virtual clock forward.
 Some important timestamps are however hidden as defaults, as listed below.
-These values can be modified using kwargs in the ``scheduler.start(...)`` call:
+These values can be modified using `kwargs` in the ``scheduler.start(...)`` call:
 
 1. ``created`` [100]: When is the observable created. 
-   That is when the ``create`` function seen in the basic example.
+   That is when the ``create`` function seen in the basic example is called.
 2. ``subscribed`` [200]: When does the subscription occur. 
    This explains the above emission timestamps: 
    consider the first emission @500; given that we are using a cold observable,
-   and subscribe to it at 200, the "source"'s timeline starts at 200 and only 300 ticks later, it emits.
+   and subscribe to it at 200, the `source`'s timeline starts at 200 and only 300 ticks later, it emits.
 3. ``disposed`` [1000]: When the subscription is disposed
 
-Keep the following in mind when modifying these values:
+Gotchas when modifying these values:
 
-1. Do not use `0` as values since the code ignores that
-2. If you change ``subscribed`` to be lower than 100, you need to change ``created`` as well
+1. Do not use `0` as values for created/subscribed since the code would ignore it.
+2. If you change ``subscribed`` to be lower than 100, you need to change ``created`` as well,
    otherwise nothing will happen.
 
 An alternative using marbles
@@ -134,13 +132,17 @@ There is a simplified flow available in `reactivex.testing.marbles` and here's a
             assert results == outcome
 
 This method makes for very quick to write, and easy to read, tests.
+At this moment however, it does not allow for testing subscriptions.
 
 
 Testing an observable factory
 .............................
 
-An observable created from `Observable(subscribe)` can be just as easily tested. 
-Let's use this example to additionally test a Disposable case.
+An observable created directly from :class:`Observable <reactivex.Observable>` 
+can be just as easily tested.
+
+In this example, we will additionally test a case where a 
+:class:`Disposable <reactivex.Disposable>` is used.
 
 .. code:: python
 
@@ -163,7 +165,7 @@ Let's use this example to additionally test a Disposable case.
             on_next(220, 0),
             on_completed(220)
         ]
-        assert a == 43
+        assert a == 43  # shows that our Disposable's action was as expected
 
 
 Testing errors
@@ -188,20 +190,20 @@ Let's remedy that below.
         # At times it's better not to test the exact exception, 
         # maybe its message changes with time or other reasons
         # We can test a specific notification's details as follows:
-        message, err = result.messages
-        assert message.time == 130
-        assert err.time == 230
-        assert message.value.kind == 'N'  # Notification
-        assert err.value.kind == 'E'  # E for errors
-        assert message.value.value == 1
-        assert type(err.value.exception) == ValueError  # look at .exception for errors
+        first_notification, error_notification = result.messages
+        assert first_notification.time == 130
+        assert error_notification.time == 230
+        assert first_notification.value.kind == 'N'  # Notification
+        assert error_notification.value.kind == 'E'  # E for errors
+        assert first_notification.value.value == 1
+        assert type(error_notification.value.exception) == ValueError  # look at .exception for errors
 
 
 Testing subscriptions, multiple observables, hot observables
 ............................................................
 
 ``scheduler.start`` only allows for a single subscription. 
-Some cases like e.g. `operators.partition` require more.
+Some cases like e.g. ``operators.partition`` require more.
 The examples below showcase some less commonly needed testing tools.
 
 .. code:: python
@@ -218,7 +220,9 @@ The examples below showcase some less commonly needed testing tools.
         even.subscribe(steven)
         odd.subscribe(todd)
 
-        # Note! Since it's not "start" which creates the subscription, they actually occur at t=0
+        # Note! Since the subscription is not created within 
+        # `scheduler.start` below, the usual `subscribed` delay of t=200
+        # is not in effect. The subscriptions therefore occur at t=0
         scheduler.start()
 
         assert steven.messages == [
@@ -242,20 +246,23 @@ The examples below showcase some less commonly needed testing tools.
         shared = source.pipe(
             operators.share()
         )
-        """first sub"""
+        # Creating our story:
+        # first sub is set to occur at t=200; this creates a sub on source
         scheduler.schedule_relative(200, lambda *_: subs.append(shared.subscribe(scheduler=scheduler)))
-        # second sub, should not sub to source itself
+        # second sub does not create a new sub on source, due to the `share` operator
         scheduler.schedule_relative(300, lambda *_: subs.append(shared.subscribe(scheduler=scheduler)))
+        # second sub ends
         scheduler.schedule_relative(500, lambda *_: subs[1].dispose())
+        # first sub ends… and since there is no sub remaining, the only sub on source should be disposed too
         scheduler.schedule_relative(600, lambda *_: subs[0].dispose())
-        """end first sub"""
-        # no existing sub should sub again onto source - we never dispose of it
+        # no existing sub on source, therefore this will create a new one
+        # we never dispose of it; we will test that infinite sub in the assertions
         scheduler.schedule_relative(900, lambda *_: subs.append(shared.subscribe(scheduler=scheduler)))
 
         scheduler.start()
         # Check that the submissions on the source are as expected
         assert source.subscriptions == [
-            Subscription(200, 600),
+            Subscription(200, 600), # only one sub from 200 to 600
             Subscription(900),  # represents an infinite subscription
         ]
 
@@ -279,9 +286,9 @@ The examples below showcase some less commonly needed testing tools.
         # the subscription starts at 200;
         # since `source` is a hot observable, the notification @190 will not be caught
         # the next notification is at 300 ticks, 
-        # which, on our subscription, will show at 100 ticks (300-200 from subscribed)
-        # or 5 "-" each representing 20 ticks (timespan=20 in to_marbles)
-        # then the 42 is received
-        # and then nothing for another 200 ticks, so 10 "-" before complete
+        # which, on our subscription, will show at 100 ticks (300-200 from subscription delay)
+        # or 5 "-" each representing 20 ticks (timespan=20 in `to_marbles`).
+        # Then the "42" notification is received
+        # and then nothing for another 200 ticks, which is equal to 10 "-", before complete
         assert message.value.value == '-----(42)----------|'