import com.mongodb.ConnectionString;
import com.mongodb.MongoClientSettings;
import com.mongodb.MongoException;
import com.mongodb.client.ClientSession;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import org.bson.Document;
import org.bson.types.ObjectId;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class WriteConflictTest {
    public static void main(String[] args) throws InterruptedException {
        ConnectionString connectionString = new ConnectionString("mongodb://localhost,localhost:27018");
        MongoClient client = MongoClients.create(MongoClientSettings.builder()
                .applyConnectionString(connectionString)
//                .addCommandListener(new CommandListener() {
//                    @Override
//                    public void commandStarted(final CommandStartedEvent event) {
//                        System.out.println(event.getCommand());
//                    }
//
//                    @Override
//                    public void commandSucceeded(final CommandSucceededEvent event) {
//                        System.out.println(event.getResponse());
//                        System.out.println();
//                    }
//
//                    @Override
//                    public void commandFailed(final CommandFailedEvent event) {
//                        System.out.println(event.getThrowable().getMessage());
//                        System.out.println();
//                    }
//                })
                .build());

        MongoDatabase database = client.getDatabase("test");

        String collectionName = "transactions";

        MongoCollection<Document> collection = database.getCollection(collectionName);

        Document document = new Document("_id", new ObjectId());

        setUp(database, collection, document);

        ExecutorService executor = Executors.newFixedThreadPool(2);

        executor.submit(() -> {
            replaceDoc(client, collection, document);
        });

        executor.submit(() -> {
            replaceDoc(client, collection, document);
        });

        executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
        System.out.println("EXITING");
    }

    private static void setUp(final MongoDatabase database, final MongoCollection<Document> collection, final Document document) {
        collection.drop();
        database.createCollection(collection.getNamespace().getCollectionName());

        collection.insertOne(document);
    }

    private static void replaceDoc(final MongoClient client, final MongoCollection<Document> collection, final Document filter) {
        Document replacement = new Document(filter);
        for (int i = 0; i < Integer.MAX_VALUE; i++) {
            replacement.put("mod", Thread.currentThread().getName() + ": " + i);
            try (ClientSession session = client.startSession()) {
                session.startTransaction();
                collection.replaceOne(session, filter, replacement);
                session.commitTransaction();
                System.out.println("Committed on thread" + Thread.currentThread().getName() + " with mod: " + replacement.get("mod"));
            } catch (MongoException e) {
                System.out.println("Failed to commit on thread" + Thread.currentThread().getName() + ": " + e.getMessage());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


}