The pathway the current process is producing from.
A produce checkpoint needs the pathway the message being answered was consumed on, and passing it through every function signature of a pipeline is noise. Instead the stage that consumed the message leaves the pathway here, and the produce call in the same process picks it up. The value lives in the process dictionary, so it belongs to one process: a processor stage, a batch processor and a task spawned from either do not see each other's pathway.
DdDataStreams.attach/0 writes this context for both kinds of Broadway
stage - the pathway of the message a handle_message/3 call is about to
receive, and the fan-in pathway of a handle_batch/4 call - so an application
that only publishes from those callbacks never touches this module. Set it
yourself in a stage the library does not know about:
pathway = my_stage_consume_pathway
:ok = DdDataStreams.Context.put(pathway)
attributes = Enum.into(DdDataStreams.produce(topic, data), existing_attributes)
MyApp.Publisher.publish(topic, data, attributes: attributes)The context is per stage process and is overwritten by each message, so it is
not cleared when a message is done: until the next message arrives it holds
the pathway of the last one, which is why clear/0 exists for a stage that
produces outside the message it consumed.
Summary
Functions
Removes the process's current pathway.
Returns the process's current pathway, or nil when it has none.
Returns the extractor that named the current pathway's consume edge, or nil
when this process has not consumed a message.
Returns the producer info of the message the current pathway was consumed
with, or %{producer: nil, topology: nil} when this process has no pathway
or its pathway was set without one.
Stores pathway as the process's current pathway, with the extractor that
named its consume edge.
Functions
@spec clear() :: :ok
Removes the process's current pathway.
iex> DdDataStreams.Context.put(%DdDataStreams.Pathway{hash: 7})
iex> DdDataStreams.Context.clear()
:ok
iex> DdDataStreams.Context.get()
nil@spec get() :: DdDataStreams.Pathway.t() | nil
Returns the process's current pathway, or nil when it has none.
iex> DdDataStreams.Context.put(%DdDataStreams.Pathway{hash: 7})
iex> DdDataStreams.Context.get()
%DdDataStreams.Pathway{hash: 7, pathway_start: 0, edge_start: 0, extractor: nil}@spec get_extractor() :: module() | nil
Returns the extractor that named the current pathway's consume edge, or nil
when this process has not consumed a message.
iex> DdDataStreams.Context.put(%DdDataStreams.Pathway{hash: 7}, Enum)
iex> DdDataStreams.Context.get_extractor()
Enum@spec get_producer_info() :: DdDataStreams.Extractor.producer_info()
Returns the producer info of the message the current pathway was consumed
with, or %{producer: nil, topology: nil} when this process has no pathway
or its pathway was set without one.
This is what DdDataStreams.produce/3 gives the extractor for the outgoing
edge, so a produce can report identity from the same producer configuration
the consume side saw.
iex> DdDataStreams.Context.put(%DdDataStreams.Pathway{hash: 7}, Enum, %{producer: {MyProducer, []}, topology: MyPipeline})
iex> DdDataStreams.Context.get_producer_info()
%{producer: {MyProducer, []}, topology: MyPipeline}
iex> DdDataStreams.Context.clear()
:ok
iex> DdDataStreams.Context.get_producer_info()
%{producer: nil, topology: nil}@spec put(DdDataStreams.Pathway.t()) :: :ok
Stores pathway as the process's current pathway, with the extractor that
named its consume edge.
iex> DdDataStreams.Context.put(%DdDataStreams.Pathway{hash: 7})
:okThe extractor is what a produce in this process reports the outgoing edge as,
so it follows the pathway: a pipeline that consumes from one technology and
publishes to another passes its own extractor to DdDataStreams.produce/3
instead. nil leaves no extractor behind.
@spec put(DdDataStreams.Pathway.t(), module() | nil) :: :ok
@spec put( DdDataStreams.Pathway.t(), module() | nil, DdDataStreams.Extractor.producer_info() | nil ) :: :ok