Use the stream consumer API