/*
 * Copyright Debezium Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */
package io.debezium.examples.graphql.eventsource;

import java.time.ZonedDateTime;
import java.util.Random;

import javax.persistence.EntityManager;
import javax.persistence.EntityManagerFactory;
import javax.persistence.Persistence;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Triggers change events by persisting {@link Order} records.
 *
 * @author Gunnar Morling
 */
class EventSource {

    private static final Logger LOG = LoggerFactory.getLogger(EventSource.class);

    private boolean running = true;
    private Thread thread;
    private final Random random = new Random();

    public void run() {
        thread = new Thread(() -> {
            EntityManagerFactory entityManagerFactory = Persistence.createEntityManagerFactory("templatePU");
            EntityManager entityManager = entityManagerFactory.createEntityManager();

            entityManager.getTransaction().begin();
            Object[] minMaxCustomerIds = (Object[]) entityManager.createNativeQuery("select min(id), max(id) from customers").getSingleResult();
             Object[] minMaxProductIds = (Object[]) entityManager.createNativeQuery("select min(id), max(id) from products").getSingleResult();

            entityManager.getTransaction().commit();

            int i = 0;
            while (running) {
                if (i % 50 == 0) {
                    entityManager.getTransaction().begin();
                }

                entityManager.persist(getRandomOrder(entityManager, (int)minMaxCustomerIds[0], (int)minMaxCustomerIds[1], (int)minMaxProductIds[0], (int)minMaxProductIds[1]));

                i++;
                try {
                    Thread.sleep(50);
                }
                catch (InterruptedException e) {
                    LOG.info("Interrupted");
                    running = false;
                }

                if (i % 50 == 0) {
                    LOG.info("Inserted {} orders", i);
                    entityManager.getTransaction().commit();
                    entityManager.clear();
                }
            }

            LOG.info("Clean-up");

            entityManager.close();
            entityManagerFactory.close();
        });

        thread.start();
    }

    private Order getRandomOrder(EntityManager entityManager, int minCustomerId, int maxCustomerId, int minProductId, int maxProductId) {
        int customerId = minCustomerId + random.nextInt(maxCustomerId - minCustomerId + 1);
        int productId = minProductId + random.nextInt(maxProductId - minProductId + 1);
        int quantity = random.nextInt(4) + 1;

        return new Order(
                ZonedDateTime.now(),
                customerId,
                productId,
                quantity);
    }

    public void stop() {
        try {
            thread.interrupt();
            thread.join();
        }
        catch (InterruptedException e) {
        }
    }
}
@@ -1216,33 +1216,32 @@
 ();%0A            
- 
 Object%5B%5D minMaxP
@@ -1449,57 +1449,80 @@
     
-while (running) %7B%0A                if (i %25 50 == 0
+long timestamp = System.currentTimeMillis();%0A%0A            while (running
 ) %7B%0A
@@ -1529,36 +1529,32 @@
                 
-    
 entityManager.ge
@@ -1577,35 +1577,16 @@
 egin();%0A
-                %7D%0A%0A
         
@@ -1615,16 +1615,41 @@
 persist(
+%0A                        
 getRando
@@ -1671,24 +1671,25 @@
 nager, (int)
+ 
 minMaxCustom
@@ -1699,24 +1699,25 @@
 ds%5B0%5D, (int)
+ 
 minMaxCustom
@@ -1723,32 +1723,33 @@
 merIds%5B1%5D, (int)
+ 
 minMaxProductIds
@@ -1758,16 +1758,17 @@
 %5D, (int)
+ 
 minMaxPr
@@ -1782,17 +1782,16 @@
 s%5B1%5D));%0A
-%0A
         
@@ -1802,220 +1802,372 @@
     
-i++;%0A                try %7B%0A                    Thread.sleep(50);%0A                %7D%0A                catch (InterruptedException e) %7B%0A                    LOG.info(%22Interrupted%22);%0A                    running = false
+entityManager.getTransaction().commit();%0A                entityManager.clear();%0A%0A                i++;%0A                if (i %25 50 == 0) %7B%0A                    long newTimestamp = System.currentTimeMillis();%0A                    LOG.info(%22Inserted 50 orders in %7B%7D ms (total orders now: %7B%7D)%22, (newTimestamp - timestamp), i);%0A                    timestamp = newTimestamp
 ;%0A  
@@ -2203,81 +2203,153 @@
     
-if (i %25 50 == 0) %7B%0A                    LOG.info(%22Inserted %7B%7D orders%22, i);
+try %7B%0A                    int x = random.nextInt(50) + 5;%0A                    Thread.sleep(x);%0A                %7D catch (InterruptedException e) %7B
 %0A   
@@ -2369,46 +2369,30 @@
     
-entityManager.getTransaction().commit(
+LOG.info(%22Interrupted%22
 );%0A 
@@ -2410,37 +2410,31 @@
         
-entityManager.clear()
+running = false
 ;%0A      
@@ -3240,24 +3240,16 @@
        %7D
-%0A       
  catch (

Change temporal distribution of Order creation