@@ -686,11 +686,17 @@ parReplicateM cfg n = parSequence cfg . Stream.replicate n
686-------------------------------------------------------------------------------
686-------------------------------------------------------------------------------
688-- Note: we can use another API with two callbacks stop and yield if we want
688-- Note: we can use another API with two callbacks stop and yield if we want
689−-- the callback to be able to indicate end of stream.
689+-- the callback to be able to indicate end of stream. Or we can use a Maybe
690+-- stream where Nothing indicates end of stream.
692+-- XXX Rename to parNewCallback
694+-- | Returns an entangled pair of a callback and a stream i.e. whenever the
695+-- callback is called a value appears in the stream. The stream is infinite,
696+-- there is no way for the callback to indicate that it is done now.
691−-- | Generates a callback and a stream pair. The callback returned is used to
698+-- The callback queues a value to a concurrent channel associated with the
692−-- queue values to the stream. The stream is infinite, there is no way for the
699+-- stream. The stream can be evaluated safely in any thread.
693−-- callback to indicate that it is done now.
695-- /Pre-release/
701-- /Pre-release/
@@ -701,10 +707,13 @@ newCallbackStream = do
702 -- XXX Add our own thread-id to the SVar as we can not know the callback's
708 -- XXX Add our own thread-id to the SVar as we can not know the callback's
703 -- thread-id and the callback is not run in a managed worker. We need to
709 -- thread-id and the callback is not run in a managed worker. We need to
704− -- handle this better.
710+ -- handle this better. The caller thread might be killed by the Channel if
711+ -- the stream evaluator dies.
705 liftIO myThreadId
713 liftIO myThreadId
706 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
714 >>= modifyThread (workerThreads chan) (outputDoorBell chan)
716+ -- XXX We can use a "Maybe a" here. Use Nothing to send a Stop event.
708 let callback a =
717 let callback a =
@@ -713,9 +722,22 @@ newCallbackStream = do
713 -- XXX Use fromChannelD?
722 -- XXX Use fromChannelD?
714 return (callback, fromChannel chan)
723 return (callback, fromChannel chan)
716−-- | Supplies a stream generating callback to a callback setter function. Each
725+-- XXX Rename this to parSetCallback. Also take the Channel config as argument.
717−-- invocation of the callback results in a value being generated in the
726+-- What config can be set by user here?
718−-- resulting stream.
727+--
728+-- XXX What happens if an exception occurs when evaluating the stream? The
729+-- result of callback can be used to communicate that. But we can only know
730+-- about the exception on the next callback call. For better handling the user
731+-- can supply an exception sender function as argument to fromCallback.
733+-- | @fromCallback f@ creates an entangled pair of a callback and a stream i.e.
734+-- whenever the callback is called a value appears in the stream. The function
735+-- @f@ is invoked with the callback as argument, and the stream is returned.
736+-- @f@ would store the callback for calling it later for generating values in
739+-- The callback queues a value to a concurrent channel associated with the
740+-- stream. The stream can be evaluated safely in any thread.
720-- /Pre-release/
742-- /Pre-release/