package de.springbootbuch.reactive.watchednow;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import static org.springframework.web.reactive.function.server.RequestPredicates.POST;
import org.springframework.web.reactive.function.server.RouterFunction;
import static org.springframework.web.reactive.function.server.RouterFunctions.route;
import static org.springframework.web.reactive.function.server.ServerResponse.ok;

/**
 * Part of springbootbuch.de.
 *
 * @author Michael J. Simons
 * @author @rotnroll666
 */
@SpringBootApplication
public class Application {

	public static void main(String[] args) {
		SpringApplication.run(Application.class, args);
	}

	@Bean
	RouterFunction<?> routerFunction(final FilmWatchedEventRepository filmWatchedEventRepository) {
		return route(
			POST("/api/filmWatched"), request -> 
			ok().body(
				request.bodyToMono(Film.class)
					.map(FilmWatchedEvent::new)
					.flatMap(filmWatchedEventRepository::save), 
				FilmWatchedEvent.class
			)
		);
	}
}

@@ -41,16 +41,99 @@
 ednow;%0A%0A
+import java.time.LocalDateTime;%0Aimport org.springframework.boot.ApplicationRunner;%0A
 import o
@@ -296,16 +296,320 @@
 n.Bean;%0A
+import org.springframework.data.mongodb.core.CollectionOptions;%0Aimport org.springframework.data.mongodb.core.MongoOperations;%0Aimport org.springframework.http.MediaType;%0Aimport org.springframework.stereotype.Component;%0Aimport static org.springframework.web.reactive.function.server.RequestPredicates.GET;%0A
 import s
@@ -845,16 +845,159 @@
 .route;%0A
+import org.springframework.web.reactive.function.server.ServerRequest;%0Aimport org.springframework.web.reactive.function.server.ServerResponse;%0A
 import s
@@ -1069,16 +1069,52 @@
 onse.ok;
+%0Aimport reactor.core.publisher.Mono;
 %0A%0A/**%0A *
@@ -1399,125 +1399,804 @@
 nal 
-FilmWatchedEventRepository filmWatchedEventRepository) %7B%0A%09%09return route(%0A%09%09%09POST(%22/api/filmWatched%22), request -%3E %0A%09%09%09
+Handler handler) %7B%0A%09%09return route(POST(%22/api/filmWatched%22), handler::filmWatched)%0A%09%09%09.andRoute(GET(%22/api/watchedRightNow%22), handler::watchedRightNow);%0A%09%7D%0A%09%0A%09@Bean%0A%09ApplicationRunner applicationRunner(MongoOperations operations) %7B%0A%09%09return args -%3E %7B%0A%09%09%09if (operations.collectionExists(FilmWatchedEvent.class)) %7B%0A%09%09%09%09operations.dropCollection(FilmWatchedEvent.class);%0A%09%09%09%7D%0A%09%09%09operations.createCollection(%0A%09%09%09%09FilmWatchedEvent.class, %0A%09%09%09%09new CollectionOptions(1024 * 1024, 100, true));%0A%09%09%7D;%0A%09%7D%0A%7D%0A%0A@Component%0Aclass Handler %7B%0A%0A%09private final FilmWatchedEventRepository filmWatchedEventRepository;%0A%0A%09Handler(FilmWatchedEventRepository filmWatchedEventRepository) %7B%0A%09%09this.filmWatchedEventRepository = filmWatchedEventRepository;%0A%09%7D%0A%0A%09Mono%3CServerResponse%3E filmWatched(ServerRequest request) %7B%0A%09%09return 
 ok()
@@ -2202,17 +2202,16 @@
 ).body(%0A
-%09
 %09%09%09reque
@@ -2240,17 +2240,16 @@
 ss)%0A%09%09%09%09
-%09
 .map(Fil
@@ -2268,17 +2268,16 @@
 t::new)%0A
-%09
 %09%09%09%09.fla
@@ -2319,41 +2319,354 @@
 ve),
- %0A%09%09%09%09FilmWatchedEvent.class%0A%09%09%09)
+%0A%09%09%09FilmWatchedEvent.class%0A%09%09);%0A%09%7D%0A%09%0A%09Mono%3CServerResponse%3E watchedRightNow(ServerRequest request) %7B%0A%09%09return ok()%0A%09%09%09.contentType(MediaType.TEXT_EVENT_STREAM)%0A%09%09%09.body(%0A%09%09%09%09filmWatchedEventRepository%0A%09%09%09%09%09.streamAllBy()%0A%09%09%09%09%09.filter(e -%3E e.getWatchedOn()%0A%09%09%09%09%09%09.isAfter(LocalDateTime.now().minusSeconds(10))%0A%09%09%09%09%09)%0A%09%09%09, %0A%09%09%09FilmWatchedEvent.class
 %0A%09%09)

Add example for streamed events