import com.mongodb.client.model.Aggregates;
import com.mongodb.client.model.Projections;
import com.mongodb.client.model.changestream.ChangeStreamDocument;
import com.mongodb.client.model.changestream.FullDocument;
import com.mongodb.reactivestreams.client.ChangeStreamPublisher;
import com.mongodb.reactivestreams.client.MongoClient;
import com.mongodb.reactivestreams.client.MongoClients;
import org.bson.BsonDocument;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.Collections;
import java.util.List;

public class Main {

    public static void main(String[] args) throws InterruptedException {
        MongoClient reactiveMongoClient = MongoClients.create("mongodb://localhost:27017");

        Bson projection = Projections.include("operationType", "documentKey", "fullDocument", "updateDescription");
        List<? extends Bson> pipeline = Collections.singletonList(Aggregates.project(projection));
        final ChangeStreamPublisher<Document> publisher = reactiveMongoClient.getDatabase("test")
                .getCollection("collectionTest")
                .watch(pipeline)
                .fullDocument(FullDocument.UPDATE_LOOKUP);

        publisher.subscribe(new Subscriber<ChangeStreamDocument<Document>>() {
            @Override
            public void onSubscribe(final Subscription s) {
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(ChangeStreamDocument<Document> documentChangeStreamDocument) {
                onNextEvent(documentChangeStreamDocument);
            }

            @Override
            public void onError(final Throwable t) {
                System.out.println("Failed");
                System.out.println(t);
            }

            @Override
            public void onComplete() {
                System.out.println("Completed");
            }
        });

        Thread.sleep(Integer.MAX_VALUE);
    }

    private static void onNextEvent(ChangeStreamDocument<Document> changeStreamDocument) {
        switch (changeStreamDocument.getOperationType()) {
            case INSERT:
            case UPDATE:
                processInsertOrUpdateEvent(changeStreamDocument);
                break;
            case DELETE:
            case REPLACE:
                processDeleteOrReplaceEvent(changeStreamDocument);
                break;
            // For change streams opened up against a collection,
            // a drop event, rename event, or dropDatabase event
            // that affects the watched collection leads to an invalidate event.
            case INVALIDATE:
                break;
            default:
                break;
        }
    }


    private static void processInsertOrUpdateEvent(ChangeStreamDocument<Document> changeStreamDocument) {
        System.out.println("Insert or update event");
    }

    private static void processDeleteOrReplaceEvent(ChangeStreamDocument<Document> changeStreamDocument) {
        System.out.println("Delete or replace event");
    }
}