NonRepeatableRead.java

package io.gitrebase;

import com.mongodb.client.*;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.Updates;
import org.bson.Document;

public class NonRepeatableRead {

    private static final String DATABASE_NAME = "gitrebase";
    private static final String COLLECTION_PRODUCTS = "products";
    private static final String COLLECTION_ORDERS = "orders";

    public static void main(String[] args) throws InterruptedException {
        MongoClient mongoClient = MongoClients.create("mongodb://localhost:27017,localhost:27018,localhost:27019/?replicaSet=rs0");
        MongoDatabase database = mongoClient.getDatabase(DATABASE_NAME);
        MongoCollection<Document> products = database.getCollection(COLLECTION_PRODUCTS);
        MongoCollection<Document> orders = database.getCollection(COLLECTION_ORDERS);

        products.deleteMany(Filters.eq("category", "PIZZA"));
        Document pizza = new Document("_id", "PIZZA_001")
                .append("name", "Cheese Burst Pizza")
                .append("category", "PIZZA")
                .append("price", 350);
        products.insertOne(pizza);
        System.out.println("Inserted product: " + pizza.toJson());

        // Client A 
        Thread clientAThread = new Thread(() -> {
            try {
                // t1: Fetch product price
                System.out.println("Client A: Fetching product ...");
                Document firstRead = products.find(Filters.eq("_id", "PIZZA_001")).first();
                System.out.println("Client A : " + firstRead.toJson());

                // Simulate delay before placing order
                Thread.sleep(1000);

                // t3: Place an order with the price fetched at t1
                System.out.println("Client A: Placing order ...");
                orders.insertOne(new Document("orderId", "ORD_001")
                        .append("productId", "PIZZA_001")
                        .append("orderedPrice", firstRead.getInteger("price")));
                System.out.println("Client A: Order placed at t3.");

                // Simulate delay before fetching price again
                Thread.sleep(2000);

                // Fetch product price again
                System.out.println("Client A: Fetching product ...");
                Document secondRead = products.find(Filters.eq("_id", "PIZZA_001")).first();
                System.out.println("Client A : " + secondRead.toJson());

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        Thread clientBThread = new Thread(() -> {
            try {
                // Increment pizza price by 10%
                Thread.sleep(1000);  // ensure it happens after t1
                System.out.println("Client B: Incrementing pizza price by 10% ...");
                products.updateMany(Filters.eq("category", "PIZZA"), Updates.mul("price", 1.10));
                System.out.println("Client B: Price updated.");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        clientAThread.start();
        clientBThread.start();

        clientAThread.join();
        clientBThread.join();
    }
}