Begin -> consume -> correlate/respond -> endEndpoint handlers
Source and sink endpoint contracts for Go. The exact generated signature follows connector type, endpoint direction, protocol mode, input type, result type, and error type.
When to use
Use this page while implementing a generated business function or endpoint handler.
Behavior
- A source endpoint converts transport input into graph values. A sink endpoint converts graph values into transport requests and optionally emits responses back into the graph.
- Begin creates state, consume performs the protocol conversion, result correlation handles asynchronous graph responses, and end releases resources.
- When an Input and Sink share a request/response endpoint, generated soft associations coordinate the protocol boundary but do not replace persisted stream links.
Result correlation
A source can emit work into the graph and wait for a later result. Register the callback under the same message ID returned by the handler correlation method. The framework uses the stream context to route the result back to the pending request.
Workflow and activity endpoints
Where Temporal is supported, the endpoint configuration determines whether generated code binds the function as a workflow or an activity. This is endpoint behavior; it does not introduce an additional stream type.
Source lifecycle
Common lifecycle phases; names below are normalized for readability.
concurrencyKafka sourceLimits concurrently processed records. 0 means no handler-level limit; Kafka partitioning still bounds work.
begin requestSource lifecycleCreates request-local state and may return an updated message context before any input is emitted.
consume messageSource lifecycleDecodes the transport request or record, emits values through the source stream context, and optionally registers result correlation.
get message IDSource correlationReturns the stable key used to route a downstream result back to the request that emitted the input. Return an empty value when no result routing is used.
end requestSource lifecycleRuns after completion or failure and releases request-local resources. It receives the first lifecycle error.
EOFStreaming gRPC sourceSignals that the client has finished sending request messages while the RPC may still be waiting for graph results.
Go signatures
Concrete methods and helpers exposed by this runtime.
BeginRequest(ctx, sc) (context.Context, State, error)Go sourceStarts one inbound request or record. A non-nil error rejects it before ConsumeMessage and EndRequest.
ConsumeMessage(ctx, sc, state, transportValue, resultCtx) errorGo sourceDecodes and emits input. gRPC variants also receive a typed Sender.
GetMessageID(ctx, sc, state, result) stringGo sourceCorrelates asynchronous graph results with callbacks registered on ResultContext.
EndRequest(ctx, sc, err, state)Go sourceFinalizes source processing. The source form does not return a second error.
ResultContext.SetResultCallback(id, callback)Go source helperRegisters the callback invoked when the result stream produces a value with the same message ID.
ResultContext.Done()Go source helperSignals that synchronous input production is finished; call it only when the request may complete.
BeginRequest / ConsumeMessage / HandleResponse / EndRequestGo sinkImplements outbound HTTP, gRPC, Kafka, local, or Temporal behavior using transport-specific request and response values.
Sink lifecycle
Outbound lifecycle phases used when supported by the selected connector.
begin requestSink lifecycleCreates state for one outbound interaction and may enrich the message context.
consume messageSink lifecycleMaps a graph value to the transport request, sends it, or writes a Kafka record.
handle responseHTTP or gRPC sinkConverts a transport response to the sink result stream and can emit a typed error.
get stream IDKafka sink correlationSelects the message context or correlation identifier attached to the outbound record when the generated signature requires it.
end requestSink lifecycleFinalizes the interaction after success or failure and releases request-local resources.
Shared endpoint types
Roles repeated across transport-specific generated signatures.
Handler statePer request or recordState returned by the begin method and passed to every later lifecycle method. Use it instead of storing request-specific data on the singleton handler.
Source stream contextGraph boundaryEmits decoded input values and typed endpoint errors into the graph.
Sink stream contextGraph boundaryEmits decoded transport responses and typed endpoint errors from an outbound endpoint.
Result contextAsync correlationRegisters callbacks by message ID and marks synchronous request production complete.
SenderStreaming transportSends protocol responses or requests without exposing the connector implementation.
Transport-specific values
Connector adapters currently exposed by this framework.
HTTP source`HandlerData`, `ResultContext`Read request data, emit decoded values, and write or correlate the eventual response.
gRPC source`Sender`, `ResultContext`Supports unary and streaming RPC shapes. Streaming handlers also receive end-of-input lifecycle events.
Kafka source`ConsumerMessage`, `ResultContext`Read headers/key/value and explicitly mark or commit the record after the desired processing point.
HTTP/gRPC/Kafka sinks`Sender` and transport result contextsBuild outbound calls and convert responses or delivery outcomes into graph results.
Temporal endpoint`EndpointHandler`An endpoint is generated according to its configured Temporal role, including workflow and activity integration.
Cron endpoint`ScheduleEndpointFunction<T>`A maker creates one long-lived callback; OnTrigger receives portable schedule metadata and collects values into the Input stream.
Generated extension pattern
resultCtx.SetResultCallback(messageID, func(ctx context.Context, sc StreamContext, state State, value Result) bool {
resultCtx.Done()
return true
})
sc.Collect(ctx, decodedValue)