Merge pull request #89 from NuclearMissile/parallel_bug

Parallel bug - the behavior of parallel() and Stream.parallel_of() not same.

Author: garlontas

pystreamapi/__stream_converter.py

@@ -1,3 +1,4 @@
+# pylint: disable=protected-access
 from pystreamapi._streams.__base_stream import BaseStream
 from pystreamapi._streams.__parallel_stream import ParallelStream
 from pystreamapi._streams.__sequential_stream import SequentialStream
@@ -16,15 +17,18 @@ def to_numeric_stream(stream: BaseStream) -> NumericBaseStream:
             stream.__class__ = SequentialNumericStream
         if isinstance(stream, ParallelStream):
             stream.__class__ = ParallelNumericStream
+            stream._init_parallelizer()
         return stream
 
     @staticmethod
     def to_parallel_stream(stream: BaseStream) -> ParallelStream:
         """Converts a stream to a parallel stream."""
         if isinstance(stream, SequentialNumericStream):
             stream.__class__ = ParallelNumericStream
+            stream._init_parallelizer()
         elif isinstance(stream, SequentialStream):
             stream.__class__ = ParallelStream
+            stream._init_parallelizer()
         return stream
 
     @staticmethod

pystreamapi/_streams/__parallel_stream.py

@@ -20,6 +20,9 @@ def __init__(self, source: Iterable[stream.K]):
         super().__init__(source)
         self._parallelizer = Parallelizer()
 
+    def _init_parallelizer(self):
+        self._parallelizer = Parallelizer()
+
     @terminal
     def all_match(self, predicate: Callable[[Any], bool]):
         return all(Parallel(n_jobs=-1, prefer="threads", handler=self)

tests/_streams/test_stream_converter.py

@@ -1,5 +1,7 @@
 from unittest import TestCase
 
+from parameterized import parameterized
+
 from pystreamapi._streams.__parallel_stream import ParallelStream
 from pystreamapi._streams.__sequential_stream import SequentialStream
 from pystreamapi._streams.numeric.__parallel_numeric_stream import ParallelNumericStream
@@ -51,3 +53,10 @@ def test_convert_to_sequential_stream_parallel(self):
     def test_convert_to_sequential_stream_parallel_numeric(self):
         stream = ParallelNumericStream(["1", "2", "3"]).sequential()
         self.assertIsInstance(stream, SequentialNumericStream)
+
+    @parameterized.expand([("sequential stream", SequentialStream),
+                           ("sequential numeric stream", SequentialNumericStream)])
+    def test_convert_sequential_to_parallel_parallelizer_working(self, _, stream):
+        res = []
+        stream([1, 2, 3]).parallel().filter(lambda x: x > 1).for_each(res.append)
+        self.assertEqual(res, [2, 3])