Name: Cortex.Streams.Files
Author: Buildersoft,EnesHoxha

Cortex.Streams.Files v2.2.0 — nugetz

Cortex.Streams: Core streaming capabilities for building data pipelines.
Cortex.Streams.Kafka: Integration with Apache Kafka for robust data streaming.
Cortex.Streams.Pulsar: Integration with Apache Pulsar for versatile messaging needs.
Cortex.Streams.RabbitMQ: Integration with RabbitMQ for versatile messaging needs.
Cortex.Streams.AWSSQS: Integration with Amazon SQS for messaging needs in the cloud.
Cortex.Streams.AzureServiceBus: Integration with Azure Messaging Service Bus for messaging needs in the cloud.
Cortex.Streams.AzureBlobStorage: Integration with Azure Blob Storage for sinking messages.
Cortex.Streams.S3: Integration with AWS S3 for sinking messages.
Cortex.Streams.Files: Implementation of File Source and Sink operators.
Cortex.Streams.Http: Implementation of Http Source and Sink operators.
Cortex.Streams.Elasticsearch: Implementation of Elasticsearch Sink operators.
Cortex.Streams.CDC.MSSqlServer: CDC adapter for Microsoft Sql Server.
Cortex.Streams.CDC.PostgreSQL: CDC adapter for PostgreSQL.
Cortex.Streams.CDC.MongoDb: CDC adapter for MongoDb.
Cortex.States: Core state management functionalities.
Cortex.States.RocksDb: Persistent state storage using RocksDB.
Cortex.States.Cassandra: Persistent state storage using Cassandra.
Cortex.States.MongoDb: Persistent state storage using MongoDb.
Cortex.States.MSSqlServer: Persistent state storage using Microsoft Sql Server.
Cortex.States.PostgreSQL: Persistent state storage using PostgreSQL.
Cortex.States.ClickHouse: Persistent state storage using Clickhouse.
Cortex.States.SQLite: Persistent state storage using SQLite.
Cortex.Telemetry: Core library to add support for Tracing and Matrics.
Cortex.Telemetry.OpenTelemetry: Adds support for Open Telemetry.
Cortex.Types: Use complex types like OneOf, AllOf and AnyOf.
Cortex.Mediator: implementation of the Mediator pattern for .NET applications, designed to power clean, modular architectures like Vertical Slice Architecture and CQRS.
Cortex.Mediator.Behaviors.FluentValidation: implementation of the FluentValidation validation for Commands and Queries
Cortex.Vectors: is a High‑performance vector types—Dense, Sparse, and Bit—for AI.

# Install Cortex.Streams
dotnet add package Cortex.Streams

# Install Cortex.States
dotnet add package Cortex.States

# Install Cortex.Streams
Install-Package Cortex.Streams

# Install Cortex.States
Install-Package Cortex.States

var stream = StreamBuilder<int, int>.CreateNewStream("ExampleStream")
    .Map(x => x * 2)
    .Filter(x => x > 10)
    .Sink(Console.WriteLine)
    .Build();
stream.Start();

// emitting data to the stream
stream.Emit(2);

var stateStore = new RocksDbStateStore<string, int>("ExampleStateStore", "./data");
stateStore.Put("key1", 42);
Console.WriteLine(stateStore.Get("key1"));

var telemetryProvider = new OpenTelemetryProvider();
var stream = StreamBuilder<int, int>
    .CreateNewStream("TelemetryStream")
    .WithTelemetry(telemetryProvider)
    .Map(x => x * 2)
    .Sink(Console.WriteLine)
    .Build();

public class ClickEvent
{
    public string PageUrl { get; set; }
    public DateTime Timestamp { get; set; }
}

        static void Main(string[] args)
        {
            // Build the stream
            var stream = StreamBuilder<ClickEvent, ClickEvent>.CreateNewStream("ClickStream")
                .Stream()
                .Filter(e => !string.IsNullOrEmpty(e.PageUrl))
                .GroupBySilently(
                    e => e.PageUrl,                   // Key selector: group by PageUrl
                    stateStoreName: "ClickGroupStore")
                .AggregateSilently<string, int>(
                    e => e.PageUrl,             // Key selector for aggregation
                    (count, e) => count + 1,          // Aggregation function: increment count
                    stateStoreName: "ClickAggregateStore")
                .Sink(e =>
                {
                    Console.WriteLine($"Page: {e.PageUrl}");
                })
                .Build();

            // start the stream
            stream.Start();

// emit some events

var random = new Random();
var pages = new[] { "/home", "/about", "/contact", "/products" };

while (true)
{
    var page = pages[random.Next(pages.Length)];
    var click = new ClickEvent
    {
        PageUrl = page,
        Timestamp = DateTime.UtcNow
    };

    stream.Emit(click);
    Thread.Sleep(100); // Simulate click rate
}

// Access the aggregate state store data
var aggregateStore = stream.GetStateStoreByName<InMemoryStateStore<string, int>>("ClickAggregateStore");

// Access the groupby state store data
var groupByStore = stream.GetStateStoreByName<InMemoryStateStore<string, List<ClickEvent>>>("ClickGroupStore")


if (aggregateStore != null)
{
    Console.WriteLine("\nAggregated Click Counts:");
    foreach (var kvp in aggregateStore.GetAll())
    {
        Console.WriteLine($"Page: {kvp.Key}, Clicks: {kvp.Value}");
    }
}
else
{
    Console.WriteLine("Aggregate state store not found.");
}

git checkout -b feature/YourFeature

git commit -m "Add your feature"

git push origin feature/YourFeature

Buildersoft/Cortex.Streams.Filesv2.2.0

Get Started

Readme

Cortex Data Framework

Key Features

Use Cases

Features

Project Structure

Getting Started

Prerequisites

Installation

Using the .NET CLI

Using the Package Manager Console in Visual Studio

Usage

1. Creating a Stream

2. Using State Stores

3. Telemetry Integration

4. Real-Time Click Tracking

Contributing

How to Contribute

License

Sponsorship

How to Sponsor

Contact