問題タブ [cats-effect]

無限のデータ ストリームを消費するプログラムがあります。途中で、単純な合計と平均であるため、モノイドを形成するいくつかのメトリックを記録したいと思います。定期的に、これらのメトリックをどこかに書き出してクリアし、集計に戻りたいと考えています。私は本質的に持っています：

したがって、実行のほとんどはIO直接使用し、使用して持ち上げStateT.liftFます。また、特定の状況では、 への呼び出しをいくつか含めますrecordMetric。その最後に、ストリームがあります：

そして、定期的に、たとえば毎分程度、メトリックをダンプしたいので、試しました：

runそして、開始状態で呼び出すという通常のトップレベルのプログラム処理を行い、次に を呼び出しunsafeRunSyncます。

問題は、空のメトリックしか表示されないことです! 私のモノイドが暗黙的に空のメトリクスを提供していると思われますがsendStream、それがなぜなのか、またはそれを修正する方法がよくわかりません。sendMetrics代わりに、これらの呼び出しをメインストリームに「インターリーブ」できる方法があるでしょうか?

編集:これは最小限の完全な実行可能な例です:

今私がする場合：

次に、期待される結果が得られます-状態が出力に適切に蓄積されます。しかし、もしそうなら：

その後、一貫して空のリストが出力されます。部分的なリスト (約 2 要素) が出力されると予想していました。

AWS Kinesis Client Library を使用する scala アプリがあります。

私は、logstash エンコーダーで logback を使用して、アプリと KCL からのログを JSON としてフォーマットしています。

私のアプリも を使って書かれてcats.effects.IOいます。

上記のコードを実行すると、アプリからのログと KCL からのログが JSON アペンダーによって正しくフォーマットされます。

使用しようとすると問題が発生しますcats.effects.IOApp：

このバージョンを実行すると、アプリからのログは引き続き JSON アペンダーによって正しくフォーマットされますが、KCL からのログはデフォルトの基本ロガーに戻ります。

これをボンネットの下での使用に絞り込みました。ボンネットの下で本質的に行っているFiberことを使用すると、問題を再現できます。私は JVM で実行しているので、これは内部で実行されているコードです。run(args.toList).start.flatMap(_.join).unsafeRunSyncIOApp

私のlogback.xml：

これはおもちゃの Scala プログラムで、コンソールから 10 個の数字を読み取り、それぞれに 1 を加えて出力します。

上記のコードはコンパイルされません。私は得る：

私は何を間違っていますか？