title: How to make Spring Cloud Stream consumer in Webflux application?
body: <p>I have a Webflux based microservice that has a simple reactive repository:</p>
  <pre class="lang-java prettyprint-override"><code>    public interface NotificationRepository extends ReactiveMongoRepository&lt;Notification, ObjectId&gt; {
      }
  </code></pre>
  <p>Now I would like to extend this microservice to consume event messages from Kafka. This message/event will be then saved into the database.</p>
  <p>For the Kafka listener, I used Spring Cloud Stream. I created some simple Consumer and it works great - I'm able to consume the message and save it into the database.</p>
  <pre class="lang-java prettyprint-override"><code>    @Bean
      public Consumer&lt;KStream&lt;String, Event&gt;&gt; documents(NotificationRepository repository) {
          return input -&gt;
                  input.foreach((key, value) -&gt; {
                      LOG.info(&quot;Received event, Key: {}, value: {}&quot;, key, value);
                      repository.save(initNotification(value)).subscribe();
                  });
      }
  </code></pre>
  <p>But is this the correct way to connect Spring Cloud Stream consumer and reactive repository? It doesn't look like it is when I have to call <code>subscribe()</code> in the end.</p>
  <p>I read the <a href="https://spring.io/blog/2019/11/25/announcing-the-release-of-spring-cloud-stream-horsham-3-0-0-release" rel="nofollow noreferrer">Spring Cloud Stream documentation (for 3.0.0 release)</a> and they say</p>
  <pre><code>Native support for reactive programming - since v3.0.0 we no longer distribute spring-cloud-stream-reactive modules and instead relying on native reactive support provided by spring cloud function. For backward compatibility you can still bring spring-cloud-stream-reactive from previous versions.
  </code></pre>
  <p>and also in <a href="https://youtu.be/5Mgni6AYnWg?t=400" rel="nofollow noreferrer">this presentation video</a> they mention they have reactive programming support using project reactor. So I guess there is a way I just don't know it. Can you show me how to do it right?</p>
  <p>I apologize if this all sounds too stupid but I'm very new to Spring Cloud Stream and reactive programming and haven't found many articles describing this.</p>