// Path: myeslib-core/src/main/java/org/myeslib/core/data/Snapshot.java
// @SuppressWarnings("serial")
// @Value
// public class Snapshot<A extends AggregateRoot> implements Serializable {
// 	
// 	private A aggregateInstance;
// 	private Long version;
// 
// }
// 
// Path: myeslib-core/src/main/java/org/myeslib/core/storage/SnapshotReader.java
// public interface SnapshotReader<K, A extends AggregateRoot> {
// 	
// 	public Snapshot<A> get(final K id) ;
// 
// }
// 
// Path: inventory-aggregate-root/src/main/java/org/myeslib/example/SampleDomain.java
// @Data
// public static class InventoryItemAggregateRoot implements AggregateRoot {
// 
//     UUID id;
//     String description;
//     Integer available = 0;
// 
//     public void on(InventoryItemCreated event) {
//         this.id = event.id;
//         this.description = event.description;
//         this.available = 0;
//     }
// 
//     public void on(InventoryIncreased event) {
//         this.available = this.available + event.howMany;
//     }
// 
//     public void on(InventoryDecreased event) {
//         this.available = this.available - event.howMany;
//     }
//         
//     public boolean isAvailable(int howMany) {
//         return getAvailable() - howMany >= 0;
//     }
// 
// }
// 
// Path: myeslib-util/src/main/java/org/myeslib/util/hazelcast/HzJobLocker.java
// @Slf4j
// public class HzJobLocker {
// 
// 	public static final String LOCK_FAILED_MSG = "Cannot lock. May be there is another job instance working concurrently";
// 	private static final String TARGET_MAP = "distributedLockMap";
// 
// 	@Inject
// 	public HzJobLocker(HazelcastInstance instance, @Named("job.id") String jobId, @Named("job.lock.duration") String jobDuration) {
// 		this.instance = instance;
// 		this.jobId = jobId;
// 		this.jobDuration = jobDuration;
// 	}
// 
// 	private final HazelcastInstance instance;
// 	private final String jobId;
// 	private final String jobDuration;
// 	
// 	public boolean lock() {
// 		boolean success ;
// 		try {
// 			TimeUnitHelper helper = new TimeUnitHelper(jobDuration);
// 			Long durationInNumber = helper.getDurationAsNumber();
// 			TimeUnit durationTime = helper.getDurationTime();
// 			log.info(String.format("Will try to lock for jobId %s for %s %s...", jobId, durationInNumber.toString(), durationTime.toString()));
// 			IMap<String, ?> map = instance.getMap(TARGET_MAP);
// 			map.tryLock(jobId, durationInNumber, durationTime);
// 			success = map.isLocked(jobId);
// 		} catch (Throwable t) {
// 			t.printStackTrace();
// 			success = false;
// 		}
// 		return success;
// 	}
// 	
// }
// 
// Path: myeslib-util/src/main/java/org/myeslib/util/jdbi/ArTablesMetadata.java
// @Getter
// public class ArTablesMetadata {
// 	
// 	@Inject
// 	public ArTablesMetadata(@Named("aggregateRootName") String aggregateRootName) {
// 		this.aggregateRootName = aggregateRootName;
// 		this.aggregateRootTable = aggregateRootName.concat("_ar");
// 		this.unitOfWorkTable = aggregateRootName.concat("_uow");
// 	}
// 	
// 	private final String aggregateRootName;
// 	private final String aggregateRootTable;
// 	private final String unitOfWorkTable;
// 
// }


import java.math.BigDecimal;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.aggregate.UseLatestAggregationStrategy;
import org.myeslib.core.data.Snapshot;
import org.myeslib.core.storage.SnapshotReader;
import org.myeslib.example.SampleDomain.InventoryItemAggregateRoot;
import org.myeslib.util.hazelcast.HzJobLocker;
import org.myeslib.util.jdbi.ArTablesMetadata;
import org.skife.jdbi.v2.DBI;
import org.skife.jdbi.v2.Handle;
import org.skife.jdbi.v2.tweak.HandleCallback;
import org.skife.jdbi.v2.util.BigDecimalMapper;
import org.skife.jdbi.v2.util.StringMapper;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.inject.Inject;
		.routeId("timer:consume-uow")  
	  	.choice()
	  	    .when(method(jobLocker, "lock"))
	  		   .to("direct:collectEvents")
	  		.otherwise()
	  		   .log(HzJobLocker.LOCK_FAILED_MSG) 
  		.end();		
		  
	  from("direct:collectEvents")
		.routeId("direct:collectEvents")
	    .log("Starting new job") 
        .process(new GetPreviousSequenceNumber())
        .log("previousSeqNumber for this aggregateRoot ${header.previousSeqNumber}")
        .process(new GetLatestSeqNumber())
        .log("latestSeqNumber for this aggregateRoot ${header.latestSeqNumber}")
        .process(new GetIdsFromInterval())
        .choice()
        	.when(header(HOW_MANY_UOWS_FOUND).isGreaterThan(0))
        		.log("found ${header.howManyUowFound} events to synchronize")
        		.to("direct:sincronizeNewEvents")
        	.otherwise()
        		.log("did not found anything to synchronize")
        .end()
       ; 
        
      from("direct:sincronizeNewEvents")  	
        .split(body()).shareUnitOfWork()
        .streaming().parallelProcessing()
        .process(new Processor() {
			@Override
// Path: myeslib-core/src/main/java/org/myeslib/core/data/Snapshot.java
// @SuppressWarnings("serial")
// @Value
// public class Snapshot<A extends AggregateRoot> implements Serializable {
// 	
// 	private A aggregateInstance;
// 	private Long version;
// 
// }
// 
// Path: myeslib-core/src/main/java/org/myeslib/core/storage/SnapshotReader.java
// public interface SnapshotReader<K, A extends AggregateRoot> {
// 	
// 	public Snapshot<A> get(final K id) ;
// 
// }
// 
// Path: inventory-aggregate-root/src/main/java/org/myeslib/example/SampleDomain.java
// @Data
// public static class InventoryItemAggregateRoot implements AggregateRoot {
// 
//     UUID id;
//     String description;
//     Integer available = 0;
// 
//     public void on(InventoryItemCreated event) {
//         this.id = event.id;
//         this.description = event.description;
//         this.available = 0;
//     }
// 
//     public void on(InventoryIncreased event) {
//         this.available = this.available + event.howMany;
//     }
// 
//     public void on(InventoryDecreased event) {
//         this.available = this.available - event.howMany;
//     }
//         
//     public boolean isAvailable(int howMany) {
//         return getAvailable() - howMany >= 0;
//     }
// 
// }
// 
// Path: myeslib-util/src/main/java/org/myeslib/util/hazelcast/HzJobLocker.java
// @Slf4j
// public class HzJobLocker {
// 
// 	public static final String LOCK_FAILED_MSG = "Cannot lock. May be there is another job instance working concurrently";
// 	private static final String TARGET_MAP = "distributedLockMap";
// 
// 	@Inject
// 	public HzJobLocker(HazelcastInstance instance, @Named("job.id") String jobId, @Named("job.lock.duration") String jobDuration) {
// 		this.instance = instance;
// 		this.jobId = jobId;
// 		this.jobDuration = jobDuration;
// 	}
// 
// 	private final HazelcastInstance instance;
// 	private final String jobId;
// 	private final String jobDuration;
// 	
// 	public boolean lock() {
// 		boolean success ;
// 		try {
// 			TimeUnitHelper helper = new TimeUnitHelper(jobDuration);
// 			Long durationInNumber = helper.getDurationAsNumber();
// 			TimeUnit durationTime = helper.getDurationTime();
// 			log.info(String.format("Will try to lock for jobId %s for %s %s...", jobId, durationInNumber.toString(), durationTime.toString()));
// 			IMap<String, ?> map = instance.getMap(TARGET_MAP);
// 			map.tryLock(jobId, durationInNumber, durationTime);
// 			success = map.isLocked(jobId);
// 		} catch (Throwable t) {
// 			t.printStackTrace();
// 			success = false;
// 		}
// 		return success;
// 	}
// 	
// }
// 
// Path: myeslib-util/src/main/java/org/myeslib/util/jdbi/ArTablesMetadata.java
// @Getter
// public class ArTablesMetadata {
// 	
// 	@Inject
// 	public ArTablesMetadata(@Named("aggregateRootName") String aggregateRootName) {
// 		this.aggregateRootName = aggregateRootName;
// 		this.aggregateRootTable = aggregateRootName.concat("_ar");
// 		this.unitOfWorkTable = aggregateRootName.concat("_uow");
// 	}
// 	
// 	private final String aggregateRootName;
// 	private final String aggregateRootTable;
// 	private final String unitOfWorkTable;
// 
// }


// Path: inventory-jdbi/src/main/java/org/myeslib/example/jdbi/routes/JdbiConsumeEventsRoute.java
import java.math.BigDecimal;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import lombok.RequiredArgsConstructor;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.aggregate.UseLatestAggregationStrategy;
import org.myeslib.core.data.Snapshot;
import org.myeslib.core.storage.SnapshotReader;
import org.myeslib.example.SampleDomain.InventoryItemAggregateRoot;
import org.myeslib.util.hazelcast.HzJobLocker;
import org.myeslib.util.jdbi.ArTablesMetadata;
import org.skife.jdbi.v2.DBI;
import org.skife.jdbi.v2.Handle;
import org.skife.jdbi.v2.tweak.HandleCallback;
import org.skife.jdbi.v2.util.BigDecimalMapper;
import org.skife.jdbi.v2.util.StringMapper;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.inject.Inject;

		.routeId("timer:consume-uow")  
	  	.choice()
	  	    .when(method(jobLocker, "lock"))
	  		   .to("direct:collectEvents")
	  		.otherwise()
	  		   .log(HzJobLocker.LOCK_FAILED_MSG) 
  		.end();		
		  
	  from("direct:collectEvents")
		.routeId("direct:collectEvents")
	    .log("Starting new job") 
        .process(new GetPreviousSequenceNumber())
        .log("previousSeqNumber for this aggregateRoot ${header.previousSeqNumber}")
        .process(new GetLatestSeqNumber())
        .log("latestSeqNumber for this aggregateRoot ${header.latestSeqNumber}")
        .process(new GetIdsFromInterval())
        .choice()
        	.when(header(HOW_MANY_UOWS_FOUND).isGreaterThan(0))
        		.log("found ${header.howManyUowFound} events to synchronize")
        		.to("direct:sincronizeNewEvents")
        	.otherwise()
        		.log("did not found anything to synchronize")
        .end()
       ; 
        
      from("direct:sincronizeNewEvents")  	
        .split(body()).shareUnitOfWork()
        .streaming().parallelProcessing()
        .process(new Processor() {
			@Override
			public void process(Exchange e) throws Exception {