Amazon Kinesis + Reactive Extensionsによる簡易CEP

// とりあえずAWSのキーと、ストリーム名で生成する感じ
var client = new ObservableKinesisClient("awsAccessId", "awsSecretAccessKey", RegionEndpoint.APNortheast1, streamName: "KinesisTest");
 
// データの登録。オブジェクトを投げ込むとJSONシリアライズしたのを叩き込む。
await client.PutRecordAsync(new { Date = DateTime.Now, Value = "ほげほげほげほげ" });
 
// ObserveRecordDynamicでJSONのストリームとして購読できる
client.ObserveRecordDynamic()
    .Where(x => x.Value != "ほげ") // xはdynamicなのでどんなSchemaのJSONも自由に辿れる
    .Select(x => x.Date + ":" + x.Value)
    .Subscribe(Console.WriteLine);

// JSON.NET, AWSSDK, Rx-Mainの参照が必要
public class ObservableKinesisClient
{
    readonly UTF8Encoding encoding = new UTF8Encoding(false);
    readonly JsonSerializer serializer = new JsonSerializer() { Formatting = Newtonsoft.Json.Formatting.None }; // ThreadSafeだよ
    readonly string streamName;
    readonly AmazonKinesisClient kinesis; // ThreadSafeなのかは知らない（ぉぃ
 
    // コンストラクタはもっとまぢめにやりましょう
    public ObservableKinesisClient(string awsAccessId, string awsSecretAccessKey, RegionEndpoint endPoint, string streamName)
    {
        this.kinesis = new AmazonKinesisClient(awsAccessId, awsSecretAccessKey, endPoint);
        this.streamName = streamName;
    }
 
    // ようするにObjectを1レコードずつJSONで突っ込むもの
    public async Task<PutRecordResponse> PutRecordAsync(object value)
    {
        using (var ms = new MemoryStream())
        using (var sw = new StreamWriter(ms, encoding))
        using (var jw = new JsonTextWriter(sw) { Formatting = Formatting.None })
        {
            serializer.Serialize(jw, value);
            jw.Flush();
            ms.Position = 0;
 
            var request = new PutRecordRequest
            {
                StreamName = streamName,
                Data = ms,
                PartitionKey = Guid.NewGuid().ToString() // PartitionKeyは適当にランダム
            };
 
            // つまり1レコード1HTTP POSTということになる。
            // 大量に投げる際は素朴すぎてアレゲ感があるので、実際にやるときはまとめてから放り込んで
            // 取り出す側も↑の構造を前提にして取り出すよーな感じにしたほうがいーかもデスネー
            return await kinesis.PutRecordAsync(request).ConfigureAwait(false);
        }
    }
 
    // Dynamicが嫌な場合はSerialize<T>でおｋ。とりあえずこの例ではdynamicでやります。
    // Client内部で分配しちゃったほうがきっと自然にやさしい(Publish().RefCount())
    public IObservable<dynamic> ObserveRecordDynamic()
    {
        return Observable.Create<dynamic>(async (observer, cancellationToken) =>
        {
            var isRunningNextPipeline = false;
            try
            {
                // まずShard一覧を取得する
                // TODO:これを使いまわしちゃうとShardsの増減には対応してないよ！
                // 毎回DescribeStream読むのもアレだしたまに問い合わせとかがいいの？
                var describeStreamResponse = await kinesis.DescribeStreamAsync(new DescribeStreamRequest { StreamName = streamName }).ConfigureAwait(false);
                var shards = describeStreamResponse.StreamDescription.Shards;
 
                var nextIterators = new List<string>();
                foreach (var shard in shards)
                {
                    if (cancellationToken.IsCancellationRequested) return; // CancellationTokenの監視だいぢだいぢ
 
                    // ShardIteratorTypeは実際は取り出した位置を記録しておいてAFTER_SEQUENCE_NUMBERでやるか、LATESTでやるかがいーんじゃないでしょーか？
                    var shardIterator = await kinesis.GetShardIteratorAsync(new GetShardIteratorRequest
                    {
                        StreamName = streamName,
                        ShardId = shard.ShardId,
                        ShardIteratorType = ShardIteratorType.TRIM_HORIZON, // TRIM_HORIZON = 最初から, LATEST = 最新, AT_SEQUENCE_NUMBER = そこから, AFTER_SEQUENCE_NUMBER = 次から
                    }).ConfigureAwait(false);
 
                    var record = await kinesis.GetRecordsAsync(new GetRecordsRequest { ShardIterator = shardIterator.ShardIterator }).ConfigureAwait(false);
 
                    // Shardの順番で回してるので、このPushの順番は必ずしも「時系列ではない」ことにチューイ！
                    foreach (var item in record.Records)
                    {
                        PushRecord(item, observer, ref isRunningNextPipeline); // ObserverでPush!Push!Push!
                    }
 
                    nextIterators.Add(record.NextShardIterator);
                }
 
                // NextShardIteratorがある状態で無限ぐるぐる
                do
                {
                    if (cancellationToken.IsCancellationRequested) return; // ところどころCancellationTokenの監視 Part2
 
                    for (int i = 0; i < nextIterators.Count; i++)
                    {
                        if (cancellationToken.IsCancellationRequested) return; // ところどころCancellationTokenの監視 Part3
 
                        var shardIterator = nextIterators[i];
 
                        var record = await kinesis.GetRecordsAsync(new GetRecordsRequest { ShardIterator = shardIterator }).ConfigureAwait(false);
 
                        // こちらでも、やはりShardの順番で回してるので、状況によって必ずしも時系列にはならないことにチューイ！
                        foreach (var item in record.Records)
                        {
                            PushRecord(item, observer, ref isRunningNextPipeline); // ObserverでPush!Push!Push!
                        }
 
                        nextIterators[i] = record.NextShardIterator;
                    }
 
                    await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false); // 実質ポーリングなのでなんとなくDelayをちょっと入れてみる
 
                    nextIterators = nextIterators.Where(x => x != null).ToList(); // 明らかに非効率なこの実装はテキトーなんで真面目にやるなら真面目に書いてください:)
                } while (nextIterators.Any());
            }
            catch (Exception ex)
            {
                if (isRunningNextPipeline)
                {
                    throw;
                }
                else
                {
                    observer.OnError(ex);
                }
 
                return;
            }
 
            observer.OnCompleted();
        });
    }
 
    void PushRecord(Record record, IObserver<dynamic> observer, ref bool isRunningNextPipeline)
    {
        using (var sr = new StreamReader(record.Data, encoding)) // item.DataにMemoryStreamの形で1レコードが受け取れる
        using (var jr = new JsonTextReader(sr))
        {
            var obj = serializer.Deserialize(jr);
            isRunningNextPipeline = true;
            observer.OnNext(obj); // 1レコードをPush
            isRunningNextPipeline = false;
        }
    }
}

// Queries:(1)
client.ObserveRecordDynamic()
    .Select(x => new{ x.Name, x.Age })
    .Dump();
 
// Queries:(2)
client.ObserveRecordDynamic()
    .Where(x => x.Current == "Shibuya")
    .Select(x => new{ x.Name, x.Age })
    .Dump();
 
// Queries:(3)
client.ObserveRecordDynamic()
    .Buffer(TimeSpan.FromMinutes(5))
    .Select(xs => xs.GroupBy(x => x.Age).Select(x => new { Age = x.Key, Count = x.Count() }))
    .Dump();
 
// Queries:(4)
client.ObserveRecordDynamic()
    .Buffer(TimeSpan.FromMinutes(5))
    .Select(xs => xs.Max(x => x.Age))
    .Dump();
 
// Queries:(5)
client.ObserveRecordDynamic()
    .Where(x => x.Current == "Kyoto" && x.Attend[0] && x.Attend[1])
    .Buffer(TimeSpan.FromMinutes(5))
    .Select(xs => xs.GroupBy(x => x.User.Age).Select(x => new { Age = x.Key, Count = x.Count() }))
    .Dump();