+ * 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;
}
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.
+ *