@@ -54,9 +54,10 @@ import Control.Monad.Catch (MonadCatch)
5454import Control.Monad.IO.Class (MonadIO(..))
5555import Control.Monad.Trans.Class (lift)
5656import Control.Monad.Trans.State.Strict (get, put)
57−-- import Data.Hashable (Hashable)
57+import Data.Function ((&))
5858import Data.IORef (newIORef, readIORef, modifyIORef')
5959import Data.Kind (Type)
60+import Data.Maybe (isJust)
6061#if !(MIN_VERSION_base(4,11,0))
6162import Data.Semigroup (Semigroup(..))
6263#endif
@@ -70,6 +71,7 @@ import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
7071import qualified Data.List as List
7172import qualified Data.Map.Strict as Map
7273import qualified Streamly.Internal.Data.Array as Array
74+import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA
7375import qualified Streamly.Internal.Data.Fold as Fold
7476import qualified Streamly.Internal.Data.Parser as Parser
7577import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream
@@ -443,17 +445,29 @@ joinOuter :: MonadIO m =>
443445 -> SerialT m (Maybe a, Maybe b)
444446joinOuter eq s1 s =
445447 Stream.concatM $ do
446− arr <- Array.fromStream $ fmap (,False) s
447− return $ go arr <> leftOver arr
448+ inputArr <- Array.fromStream s
449+ let len = length inputArr
450+ foundArr <-
451+ Stream.fold
452+ (MA.writeN len)
453+ (Stream.fromList (Prelude.replicate len False))
454+ return $ go inputArr foundArr <> leftOver inputArr foundArr
448455
449456 where
450457
451− leftOver =
452− fmap (\(x, _) -> (Nothing, Just x))
453− . Stream.filter (not . snd)
454− . Array.toStream
455−
456− go arr = Stream.evalStateT (return False) $ do
458+ leftOver inputArr foundArr =
459+ let stream1 = IsStream.fromSerial $ Array.toStream inputArr
460+ stream2 = Stream.unfold MA.read foundArr
461+ in Stream.filter
462+ isJust
463+ ( Stream.zipWith (\x y ->
464+ if y
465+ then Nothing
466+ else Just (Nothing, Just x)
467+ ) stream1 stream2
468+ ) & Stream.catMaybes
469+
470+ go inputArr foundArr = Stream.evalStateT (return False) $ do
457471 a <- Stream.liftInner s1
458472 -- XXX should we use StreamD monad here?
459473 -- XXX Is there a better way to perform some action at the end of a loop
@@ -464,16 +478,16 @@ joinOuter eq s1 s =
464478 if r
465479 then Stream.nil
466480 else Stream.fromPure Nothing
467− (_i, b) <-
468− let stream = IsStream.fromSerial $ Array.toStream arr
481+ (i, b) <-
482+ let stream = IsStream.fromSerial $ Array.toStream inputArr
469483 in Stream.indexed $ fmap Just (Stream.liftInner stream) <> final
484+
470485 case b of
471− Just (b1, _used) ->
486+ Just b1 ->
472487 if a `eq` b1
473488 then do
474489 lift $ put True
475− -- XXX Need to use a mutable array
476− -- when (not used) $ Array.writeIndex i True
490+ MA.putIndex i True foundArr
477491 return (Just a, Just b1)
478492 else Stream.nil
479493 Nothing -> return (Just a, Nothing)