- * Both modes can be selected at runtime from the same fixed method parameter, - * so the component does not need separate parameters per output. - * - *
- * Important: This interface is supported only in the Studio DI runtime. - * - *
- * Split mode example (one source, per-record routing): - * - *
- * {@code
- *
- * @ElementListener
- * public void process(@Input Record input,
- * @Output MultiOutputIterator out) {
- * out.setIterator(
- * mySource.stream()
- * .map(r -> isValid(r)
- * ? TaggedOutput.of("MAIN", transform(r))
- * : TaggedOutput.of("REJECT", r))
- * .iterator());
- * }
- * }
- *
- *
- * - * Independent mode example (each output has its own lazy source): - * - *
- * {@code
- *
- * @AfterGroup
- * public void afterGroup(@Output MultiOutputIterator out) {
- * out.setIterator("MAIN", mainDatabase.lazyQuery());
- * out.setIterator("REJECT", errorLog.lazyRead());
- * }
- * }
- *
- *
- * @param
- * Mutually exclusive with {@link #setIterator(String, Iterator)} within one invocation.
- *
- * @param iterator the tagged iterator routing records to their named outputs
- */
- void setIterator(Iterator
- * Mutually exclusive with {@link #setIterator(Iterator)} within one invocation.
- *
- * @param outputName the output connection name (e.g. {@code "MAIN"}, {@code "REJECT"})
- * @param iterator the lazy iterator producing records for that connection
- */
- void setIterator(String outputName, Iterator
- * Supported only in the Studio DI runtime.
- *
- * @param
- * In tagged-source mode (when the component used a
- * {@link org.talend.sdk.component.api.processor.MultiOutputIterator}),
- * this peeks one record ahead from the shared tagged source and returns whether it
- * belongs to {@code connectionName}. The peeked record is buffered and consumed by
- * the next {@link #getValue(String)} call for the same connection.
- *
- *
- * In independent-iterator mode, this directly checks the per-connection iterator/queue.
- *
- *
- * Use this method in the Studio drain loop to avoid calling {@link #getValue(String)}
- * on connections that have no data:
- *
- *
- * The {@code advancer} implementation must call {@link #setPending(String, Object)} with
- * the next output-connection name and converted record value, then return {@code true}.
- * Return {@code false} when the source is exhausted.
- *
- * @param advancer the tagged-source advancer produced by a MultiOutputIterator setup
- */
- protected void setTaggedSource(final TaggedAdvancer advancer) {
- this.taggedSource = advancer;
- this.pendingOutputName = null;
- this.pendingRecord = null;
- }
-
- /**
- * Called by a {@link TaggedAdvancer} to store the next pending record.
- *
- * @param outputName the output connection name for the record
- * @param record the converted record value
- */
- protected void setPending(final String outputName, final Object record) {
- this.pendingOutputName = outputName;
- this.pendingRecord = record;
+ return connections.entrySet().stream().anyMatch(e -> e.getValue().hasNext());
}
protected String getActualName(final String name) {
return "__default__".equals(name) ? "FLOW" : name;
}
- /**
- * Represents a single output connection's data holder.
- * Supports two modes (mutually exclusive per invocation):
- *
- * Note: The {@code source} field is only used by {@code OutputsHandler}; {@code InputsHandler}
- * uses only the queue-based push mode.
- */
@RequiredArgsConstructor
static class IO
- * Uses a single shared tagged iterator that routes even-indexed records to MAIN and
- * odd-indexed records to REJECT. The drain loop uses {@code hasDataFor} to consume
- * only the connection that has a record ready — which exercises the fix in
- * {@code hasDataFor} that skips unregistered pending records instead of stalling.
- */
- @ParameterizedTest
- @EnumSource(OutputMode.class)
- void afterGroupSplitIteratorShouldRouteRecordsWithHasDataFor(OutputMode outputMode) {
- final ComponentManager manager = ComponentManager.instance();
- final Map
- * The split processor routes even-indexed records to MAIN and odd-indexed to REJECT
- * from a SINGLE shared source iterator. Without tagged-output support this would
- * require buffering all records first.
- */
- @ParameterizedTest
- @EnumSource(OutputMode.class)
- void multiOutputIteratorShouldSplitRecordsWithoutBuffering(OutputMode outputMode) {
- final ComponentManager manager = ComponentManager.instance();
- final Map{@code
- * while (outputsHandler.hasMoreData()) {
- * if (outputsHandler.hasDataFor("MAIN"))
- * mainRow = outputsHandler.getValue("MAIN");
- * if (outputsHandler.hasDataFor("REJECT"))
- * rejectRow = outputsHandler.getValue("REJECT");
- * }
- * }
- *
- * @param connectionName the output connection name to check
- * @return {@code true} if a record for this connection is immediately available
- */
- public boolean hasDataFor(final String connectionName) {
- if (taggedSource != null) {
- // Skip any pending record for an unregistered connection before checking
- while (pendingOutputName != null && !connections.containsKey(pendingOutputName)) {
- pendingOutputName = null;
- pendingRecord = null;
- if (!taggedSource.advance()) {
- return false;
- }
- }
- if (pendingOutputName != null) {
- return pendingOutputName.equals(connectionName);
- }
- // Advance, skipping records for unregistered connections
- while (taggedSource.advance()) {
- if (connections.containsKey(pendingOutputName)) {
- return pendingOutputName.equals(connectionName);
- }
- pendingOutputName = null;
- pendingRecord = null;
- }
- return false;
- }
- final IO io = connections.get(connectionName);
- return io != null && io.hasNext();
- }
-
- /**
- * Sets a tagged-source advancer for split streaming across multiple outputs.
- * Switching to tagged mode — subsequent {@link #hasMoreData()}, {@link #hasDataFor(String)},
- * and {@link #getValue(String)} operate via the advancer instead of per-connection queues.
- *
- *
- *
- * In pull mode, {@link #hasNext()} delegates directly to the source iterator's
- * {@code hasNext()} — the source iterator is expected to be idempotent and fast.
- *