Streaming Data

public class TwoTableReceiver implements DataStreamerReceiver<Tuple, Void, Tuple> {

    public TwoTableReceiver() {
    }


}

class TwoTableReceiver : IDataStreamerReceiver<IIgniteTuple, object?, object>
{
    public async ValueTask<IList<object>?> ReceiveAsync(
        IList<IIgniteTuple> page,
        object? arg,
        IDataStreamerReceiverContext context,
        CancellationToken cancellationToken)
    {
        IRecordView<IIgniteTuple> customerTable = (await context.Ignite.Tables.GetTableAsync("customers"))!.RecordBinaryView;
        IRecordView<IIgniteTuple> addressesTable = (await context.Ignite.Tables.GetTableAsync("addresses"))!.RecordBinaryView;

        foreach (IIgniteTuple sourceItem in page)
        {
            // For each source item, the receiver extracts customer and address data and upserts it into respective tables.
            var customer = new IgniteTuple
            {
                ["id"] = sourceItem["customerId"],
                ["name"] = sourceItem["customerName"],
                ["addressId"] = sourceItem["addressId"]
            };

            var address = new IgniteTuple
            {
                ["id"] = sourceItem["addressId"],
                ["street"] = sourceItem["street"],
                ["city"] = sourceItem["city"],
            };

            await customerTable.UpsertAsync(null, customer);
            await addressesTable.UpsertAsync(null, address);
        }

        return null;
    }
}

            DataStreamerReceiverDescriptor<Tuple, Void, Tuple> desc = DataStreamerReceiverDescriptor
                    .builder(TwoTableReceiver.class)
                    .units(new DeploymentUnit(DEPLOYMENT_UNIT_NAME, DEPLOYMENT_UNIT_VERSION))
                    .build();

ReceiverDescriptor<IIgniteTuple, object?, object> desc = ReceiverDescriptor.Of(new TwoTableReceiver());

            List<Tuple> sourceData = IntStream.range(1, 10)
                    .mapToObj(i -> Tuple.create()
                            .set("customerId", i)
                            .set("customerName", "Customer " + i)
                            .set("addressId", i)
                            .set("street", "Street " + i)
                            .set("city", "City " + i))
                    .collect(Collectors.toList());

            CompletableFuture<Void> streamerFut;

            RecordView<Tuple> customersTableView = client.tables().table("Customers").recordView();

            /* Extract the target table key from each source item; since the source has "customerId" but the target table uses "id", the function maps customerId to id accordingly */
            Function<Tuple, Tuple> keyFunc = sourceItem -> Tuple.create().set("id", sourceItem.intValue("customerId"));

            /* Extract the data payload sent to the receiver. In this case, we use the entire source item as the payload */
            Function<Tuple, Tuple> payloadFunc = Function.identity();

            /* Stream data using a publisher */
            try (var publisher = new SubmissionPublisher<Tuple>()) {
                streamerFut = customersTableView.streamData(
                        publisher,
                        desc,
                        keyFunc,
                        payloadFunc,
                        null, /* Optional receiver arguments */
                        null, /* Result subscriber */
                        null /* Options */
                );

                for (Tuple item : sourceData) {
                    publisher.submit(item);
                }
            }

            streamerFut.join();

IAsyncEnumerable<IIgniteTuple> sourceData = GetSourceData();

IRecordView<IIgniteTuple> customersTable = (await client.Tables.GetTableAsync("customers"))!.RecordBinaryView;

IAsyncEnumerable<object> streamerResults = customersTable.StreamDataAsync(
sourceData,
desc,
x => new IgniteTuple { ["id"] = x["customerId"] },
x => x,
null,
DataStreamerOptions.Default,
CancellationToken.None);

await foreach (object result in streamerResults)
{
// ...
}

static async IAsyncEnumerable<IIgniteTuple> GetSourceData()
{
await Task.Yield(); // Simulate async enumeration.

    for (int i = 0; i < 10; i++)
    {
        yield return new IgniteTuple
        {
            ["customerId"] = i,
            ["customerName"] = $"Customer {i}",
            ["addressId"] = i,
            ["street"] = $"Street {i}",
            ["city"] = $"City {i}"
        };
    }
}

private static class FraudDetectorReceiver implements DataStreamerReceiver<Tuple, Void, Tuple> {
    @Override
    public CompletableFuture<List<Tuple>> receive(List<Tuple> page, DataStreamerReceiverContext ctx, Void arg) {
        List<Tuple> results = new ArrayList<>(page.size());

        for (Tuple tx : page) {
            results.add(detectFraud(tx));
        }

        return CompletableFuture.completedFuture(results);
    }

    private static Tuple detectFraud(Tuple txInfo) {
        /* Simulate fraud detection processing */
        double fraudRisk = Math.random();

        /* Add result to the tuple and return */
        return txInfo.set("fraudRisk", fraudRisk);
    }
}

class FraudDetectorReceiver : IDataStreamerReceiver<IIgniteTuple, object?, IIgniteTuple>
{
    public async ValueTask<IList<IIgniteTuple>?> ReceiveAsync(
        IList<IIgniteTuple> page,
        object? arg,
        IDataStreamerReceiverContext context,
        CancellationToken cancellationToken)
    {
        var result = new List<IIgniteTuple>(page.Count);

        foreach (var tx in page)
        {
            IIgniteTuple resTuple = await DetectFraud(tx);
            result.Add(resTuple);
        }

        return result;
    }

    private static async Task<IIgniteTuple> DetectFraud(IIgniteTuple transaction)
    {
        // Simulate fraud detection logic - add a random risk score to the tuple.
        await Task.Delay(10);
        transaction["fraudRisk"] = Random.Shared.NextDouble();
        return transaction;
    }
}

        List<Tuple> sourceData = IntStream.range(1, 10)
                .mapToObj(i -> Tuple.create()
                        .set("txId", i)
                        .set("txData", "{some-json-data}"))
                .collect(Collectors.toList());

        DataStreamerReceiverDescriptor<Tuple, Void, Tuple> desc = DataStreamerReceiverDescriptor
                .builder(FraudDetectorReceiver.class)
                .build();

        CompletableFuture<Void> streamerFut;

        TableDefinition txDummyTableDef = TableDefinition.builder("tx_dummy")
                .columns(column("id", ColumnType.INTEGER))
                .primaryKey("id")
                .build();

        Table dummyTable = client.catalog().createTable(txDummyTableDef);


        Function<Tuple, Tuple> keyFunc = sourceItem -> Tuple.create().set("id", sourceItem.value("txId"));

        /* Payload function is used to extract the payload (data that goes to the receiver) from the source item.
        /* In our case, we want to use the whole source item as the payload */
        Function<Tuple, Tuple> payloadFunc = Function.identity();

        Flow.Subscriber<Tuple> resultSubscriber = new Flow.Subscriber<>() {
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                subscription.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Tuple item) {
                System.out.println("Transaction processed: " + item);
            }

            @Override
            public void onError(Throwable throwable) {
                System.err.println("Error during streaming: " + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("Streaming completed.");
            }
        };

        try (var publisher = new SubmissionPublisher<Tuple>()) {
            streamerFut = dummyTable.recordView().streamData(
                    publisher,
                    desc,
                    keyFunc,
                    payloadFunc,
                    null,
                    resultSubscriber,
                    null
            );

            for (Tuple item : sourceData) {
                publisher.submit(item);
            }
        }

        streamerFut.join();

// Source data is a list of financial transactions.
// We want to distribute this processing across the cluster, then gather and return results
IAsyncEnumerable<IIgniteTuple> data = GetSourceData();

ReceiverDescriptor<IIgniteTuple, object?, IIgniteTuple> fraudDetectorReceiverDesc = ReceiverDescriptor.Of(new FraudDetectorReceiver());

// Streaming requires a target table to partition data.
// Use a dummy table for this scenario, because we are not going to store any data.
await client.Sql.ExecuteScriptAsync("CREATE TABLE IF NOT EXISTS TX_DUMMY (ID LONG)");

ITable dummyTable = await client.Tables.GetTableAsync("TX_DUMMY");

// Source data has "txId" field, but target dummy table has "id" column, so keyFunc maps "txId" to "id".
Func<IIgniteTuple, IIgniteTuple> keyFunc = tuple => new IgniteTuple { ["id"] = tuple["txId"] };

// Payload function is used to extract the payload (data that goes to the receiver) from the source item.
// In our case, we want to use the whole source item as the payload.
Func<IIgniteTuple, IIgniteTuple> payloadFunc = tuple => tuple;

IAsyncEnumerable<IIgniteTuple> results = dummyTable.RecordBinaryView.StreamDataAsync(
data,
fraudDetectorReceiverDesc,
keyFunc,
payloadFunc,
receiverArg: null);

await foreach (IIgniteTuple processedTx in results)
{
Console.WriteLine("Transaction processed: " + processedTx);
}

async IAsyncEnumerable<IIgniteTuple> GetSourceData()
{
await Task.Yield(); // Simulate async data source.

    for (int i = 0; i < 1000; i++)
    {
        yield return new IgniteTuple
        {
            ["txId"] = i,
            ["txData"] = "{some-json-data}"
        };
    }
}