package de.springbootbuch.reactive.watchednow;

import java.time.LocalDateTime;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.data.mongodb.core.CollectionOptions;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import static org.springframework.web.reactive.function.server.RequestPredicates.GET;
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 org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import static org.springframework.web.reactive.function.server.ServerResponse.ok;
import reactor.core.publisher.Mono;

/**
 * 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 Handler handler) {
		return route(POST("/api/filmWatched"), handler::filmWatched)
			.andRoute(GET("/api/watchedRightNow"), handler::watchedRightNow);
	}
	
	@Bean
	ApplicationRunner applicationRunner(MongoOperations operations) {
		return args -> {
			if (operations.collectionExists(FilmWatchedEvent.class)) {
				operations.dropCollection(FilmWatchedEvent.class);
			}
			operations.createCollection(
				FilmWatchedEvent.class,
				CollectionOptions.empty()
					.maxDocuments(100L)
					.size(1_048_576L)
					.capped());
		};
	}
}

@Component
class Handler {

	private final FilmWatchedEventRepository filmWatchedEventRepository;

	Handler(FilmWatchedEventRepository filmWatchedEventRepository) {
		this.filmWatchedEventRepository = filmWatchedEventRepository;
	}

	Mono<ServerResponse> filmWatched(ServerRequest request) {
		return ok().body(
			request.bodyToMono(Film.class)
				.map(FilmWatchedEvent::new)
				.flatMap(filmWatchedEventRepository::save),
			FilmWatchedEvent.class
		);
	}
	
	Mono<ServerResponse> watchedRightNow(ServerRequest request) {
		return ok()
			.contentType(MediaType.TEXT_EVENT_STREAM)
			.body(
				filmWatchedEventRepository
					.streamAllBy()
					.filter(e -> e.getWatchedOn()
						.isAfter(LocalDateTime.now().minusSeconds(10))
					)
			, 
			FilmWatchedEvent.class
		);
	}
}

@@ -73,16 +73,59 @@
 teTime;%0A
+import static java.time.LocalDateTime.now;%0A
 import o
@@ -464,32 +464,39 @@
 rations;%0Aimport 
+static 
 org.springframew
@@ -517,57 +517,26 @@
 Type
-;%0Aimport org.springframework.stereotype.Component
+.TEXT_EVENT_STREAM
 ;%0Aim
@@ -1367,16 +1367,125 @@
 s);%0A%09%7D%0A%0A
+%09@Bean%0A%09Handler handler(%0A%09%09FilmWatchedEventRepository repository%0A%09) %7B%0A%09%09return new Handler(repository);%0A%09%7D%0A%09%0A
 %09@Bean%0A%09
@@ -1517,22 +1517,16 @@
 unction(
-final 
 Handler 
@@ -1545,16 +1545,21 @@
 %09return 
+%0A%09%09%09 
 route(PO
@@ -1578,24 +1578,30 @@
 mWatched%22), 
+%0A%09%09%09%09 
 handler::fil
@@ -1623,16 +1623,21 @@
 ndRoute(
+%0A%09%09%09%09
 GET(%22/ap
@@ -1657,16 +1657,21 @@
 tNow%22), 
+%0A%09%09%09%09
 handler:
@@ -2079,19 +2079,8 @@
 %0A%7D%0A%0A
-@Component%0A
 clas
@@ -2097,16 +2097,8 @@
 %7B%0A%0A%09
-private 
 fina
@@ -2126,33 +2126,17 @@
 ository 
-filmWatchedEventR
+r
 epositor
@@ -2175,33 +2175,17 @@
 ository 
-filmWatchedEventR
+r
 epositor
@@ -2200,54 +2200,22 @@
 his.
-filmWatchedEventRepository = filmWatchedEventR
+repository = r
 epos
@@ -2259,16 +2259,19 @@
 Watched(
+%0A%09%09
 ServerRe
@@ -2275,32 +2275,34 @@
 rRequest request
+%0A%09
 ) %7B%0A%09%09return ok(
@@ -2388,33 +2388,17 @@
 flatMap(
-filmWatchedEventR
+r
 epositor
@@ -2481,16 +2481,19 @@
 ightNow(
+%0A%09%09
 ServerRe
@@ -2505,16 +2505,18 @@
  request
+%0A%09
 ) %7B%0A%09%09re
@@ -2541,26 +2541,16 @@
 entType(
-MediaType.
 TEXT_EVE
@@ -2573,30 +2573,9 @@
 ody(
-%0A%09%09%09%09filmWatchedEventR
+r
 epos
@@ -2650,30 +2650,16 @@
 isAfter(
-LocalDateTime.
 now().mi

Format for book