example fixed (new stream aggr API)