Initial commit

Author: enryold

.gitignore

@@ -0,0 +1,94 @@
+# Created by .ignore support plugin (hsz.mobi)
+### JetBrains template
+# Covers JetBrains IDEs: IntelliJ, RubyMine, PhpStorm, AppCode, PyCharm, CLion
+
+*.iml
+
+## Directory-based project format:
+.idea/
+# if you remove the above rule, at least ignore the following:
+
+# User-specific stuff:
+# .idea/workspace.xml
+# .idea/tasks.xml
+# .idea/dictionaries
+
+# Sensitive or high-churn files:
+# .idea/dataSources.ids
+# .idea/dataSources.xml
+# .idea/sqlDataSources.xml
+# .idea/dynamic.xml
+# .idea/uiDesigner.xml
+
+# Gradle:
+# .idea/gradle.xml
+# .idea/libraries
+
+# Mongo Explorer plugin:
+# .idea/mongoSettings.xml
+
+## File-based project format:
+*.ipr
+*.iws
+
+## Plugin-specific files:
+
+# IntelliJ
+/build/
+/out/
+
+# mpeltonen/sbt-idea plugin
+.idea_modules/
+
+# JIRA plugin
+atlassian-ide-plugin.xml
+
+# Crashlytics plugin (for Android Studio and IntelliJ)
+com_crashlytics_export_strings.xml
+crashlytics.properties
+crashlytics-build.properties
+
+
+
+### Eclipse template
+*.pydevproject
+.metadata
+.gradle
+bin/
+tmp/
+*.tmp
+*.bak
+*.swp
+*~.nib
+local.properties
+.settings/
+.loadpath
+
+# Eclipse Core
+.project
+
+# External tool builders
+.externalToolBuilders/
+
+# Locally stored "Eclipse launch configurations"
+*.launch
+
+# CDT-specific
+.cproject
+
+# JDT-specific (Eclipse Java Development Tools)
+.classpath
+
+# Java annotation processor (APT)
+.factorypath
+
+# PDT-specific
+.buildpath
+
+# sbteclipse plugin
+.target
+
+# TeXlipse plugin
+.texlipse
+
+.DS_Store

.travis.yml

@@ -0,0 +1,15 @@
+language: java
+jdk:
+  - oraclejdk8
+
+sudo: required
+dist: trusty
+
+# cache gradle dependencies
+# https://docs.travis-ci.com/user/languages/java#Caching
+before_cache:
+  - rm -f $HOME/.gradle/caches/modules-2/modules-2.lock
+cache:
+  directories:
+  - $HOME/.gradle/caches/
+  - $HOME/.gradle/wrapper/

README.md

@@ -0,0 +1,80 @@
+# quasar-flow 
+## Handle data flows using quasar fibers and reactive-streams.
+ [![Build Status](https://travis-ci.org/fulmicotone/com.fulmicotone.qio.svg?branch=master)](https://travis-ci.org/fulmicotone/com.fulmicotone.qio)  [![](https://jitpack.io/v/fulmicotone/com.fulmicotone.qio.svg)](https://jitpack.io/#fulmicotone/com.fulmicotone.qio)
+
+
+### Goal - Build a concurrency system that
+
+- It's easy to use.
+- It's simply to read.
+- It's simply to code.
+- Uses reactive-streams channels / processors / subscriber with quasar fibers under the hood.
+- Uses FanIn/FanOut concurrency patterns.
+- Have common methods for size/byte batching with flushing timeouts.
+
+
+### Main entities:
+
+- Emitter: Entity that emit flows of objects on a channel.  
+  - An Emitter could be a broadcast emitter.
+  - An Emitter could be a routed emitter, every processor should subscribe on a particular data object property.
+  - An Emitter could have 1-N subscribers.
+
+
+- Processor: Entity that receives an Emitter's data flow.
+  - A Processor could process an Emitter data-flow with 1-N fiber/s.
+  - A Processor could transform the emitter data-flow with a transformation function.
+  - A Processor could process an Emitter data-flow with N fibers and return 1 result Emitter using FanIn pattern.
+  - A Processor could process an Emitter data-flow with N fibers and return N result Emitter using FanOut pattern.
+  - A Processor could process an Emitter data-flow batching results grouping them by size.
+  - A Processor could process an Emitter data-flow batching results grouping them by a custom user-defined accumulator.
+  
+  
+- Consumer: Entity that could receives both Emitter or Processors data-flow 
+  - A Consumer could process an Emitter/Processor data-flow with 1-N fiber/s.
+  - A Consumer could transform an Emitter/Processor data-flow with a transformation function.
+  - A Consumer could process an Emitter/Processor data-flow with N fibers and apply an user-defined task to the result.
+  - A Consumer could process an Emitter/Processor data-flow batching results grouping them by size and apply an user-defined task to the result.
+  - A Consumer could process an Emitter/Processor data-flow batching results grouping them by a custom user-defined accumulator and apply an user-defined task to the result.
+  
+ 
+
+### Examples:
+
+
+#### Emitter -> Processor -> Consumer
+```java
+ public class Main {
+ 
+     public static void main(String[] args) {
+         
+         // Emit Task that emits 10 Strings
+         IEmitterTask<String> stringEmitterTask = publisherChannel -> {
+                     for(int i = 0; i<10; i++){ publisherChannel.send("String"+i); } 
+         };
+         
+         
+         // LINEAR LAYOUT
+         QuasarFlow.newFlow()
+                         .broadcastEmitter(stringEmitterTask) // BUILD A BROADCAST EMITTER FROM TASK
+                         .addProcessor() // ADD A PROCESSOR
+                         .process() // PROCESS 
+                         .addConsumer() // ADD A CONSUMER
+                         .consume(str -> System.out.println(str)) // CONSUME WITH CONSUMER TASK
+                         .start();
+         
+         // NESTED LAYOUT
+         QuasarFlow.newFlow()
+                .broadcastEmitter(null) // BUILD A BROADCAST EMITTER FROM TASK
+                .addProcessor(p -> { // ADD A PROCESSOR
+                    p.process() // PROCESS 
+                            .addConsumer(c -> // ADD A CONSUMER
+                                    c.consume(str -> System.out.println(str))); // CONSUME WITH CONSUMER TASK
+                })
+                .start();       
+     }
+ 
+ }
+
+
+

build.gradle

@@ -0,0 +1,38 @@
+group 'it.enryold'
+
+version '0.0.1'
+
+apply plugin: 'java'
+apply plugin: 'idea'
+
+
+sourceCompatibility = 1.8
+
+
+test {
+    useJUnitPlatform()
+    maxHeapSize = "1024m"
+    jvmArgs '-javaagent:'+file('.')+"/quasar-core-0.7.10.jar"
+}
+
+
+repositories {
+    mavenCentral()
+    maven { url "http://repo.opensourceagility.com/release" }
+    maven { url "http://repo.spring.io/libs-milestone" }
+    maven { url "http://repo.spring.io/libs-snapshot" }
+}
+
+dependencies {
+    implementation group: 'com.amazonaws', name: 'aws-java-sdk-kinesis', version: '1.11.+'
+    implementation group: 'com.amazonaws', name: 'amazon-kinesis-producer', version: '0.12.+'
+    implementation group: 'com.amazonaws', name: 'amazon-kinesis-client', version: '1.9.+'
+    implementation group: 'co.paralleluniverse', name: 'quasar-core', version: '0.7.10'
+    implementation group: 'co.paralleluniverse', name: 'quasar-reactive-streams', version: '0.7.10'
+    implementation group: 'org.slf4j', name: 'slf4j-api', version: '1.7.25'
+    implementation group: 'com.google.guava', name: 'guava', version: '23.0'
+    implementation "com.evanlennick:retry4j:0.15.0"
+
+    testImplementation('org.junit.jupiter:junit-jupiter-api:5.4.2')
+    testRuntime('org.junit.jupiter:junit-jupiter-engine:5.4.2')
+}

quasar-core-0.7.10.jar

@@ -0,0 +1,2 @@
+rootProject.name = 'quasar-flow'
+

settings.gradle

@@ -0,0 +1,42 @@
+package it.enryold.quasarflow;
+
+import it.enryold.quasarflow.interfaces.IEmitter;
+import it.enryold.quasarflow.interfaces.IEmitterTask;
+import it.enryold.quasarflow.interfaces.IRoutingKeyExtractor;
+import it.enryold.quasarflow.models.QEmitter;
+import it.enryold.quasarflow.models.QFlow;
+
+public class QuasarFlow {
+
+
+    private QFlow qFlow;
+
+    private QuasarFlow(){
+        qFlow = new QFlow();
+    }
+
+    public static QuasarFlow newFlow(){
+        return new QuasarFlow();
+    }
+
+    public <T, E extends IEmitter<T>> E broadcastEmitter(IEmitterTask<T> task){
+        return new QEmitter<T>(qFlow)
+                .broadcastEmitter(task);
+    }
+
+    public <T, E extends IEmitter<T>> E broadcastEmitter(IEmitterTask<T> task, String name){
+        return new QEmitter<T>(qFlow, name)
+                .broadcastEmitter(task);
+    }
+
+    public <T, E extends IEmitter<T>> E routedEmitter(IEmitterTask<T> task, IRoutingKeyExtractor<T> routingKeyExtractor){
+        return new QEmitter<T>(qFlow)
+                .routedEmitter(task, routingKeyExtractor);
+    }
+
+    public <T, E extends IEmitter<T>> E routedEmitter(IEmitterTask<T> task, String name, IRoutingKeyExtractor<T> routingKeyExtractor){
+        return new QEmitter<T>(qFlow, name)
+                .routedEmitter(task, routingKeyExtractor);
+    }
+
+}

src/main/java/it/enryold/quasarflow/QuasarFlow.java

@@ -0,0 +1,46 @@
+package it.enryold.quasarflow.abstracts;
+
+import it.enryold.quasarflow.components.IAccumulator;
+import it.enryold.quasarflow.components.IAccumulatorLengthFunction;
+import com.google.common.util.concurrent.AtomicDouble;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public abstract class AbstractAccumulator<E, T> implements IAccumulator<E, T> {
+
+    private double byteSizeLimit;
+    protected List<E> accumulator = new ArrayList<>();
+    private AtomicDouble accumulatorSize = new AtomicDouble(0);
+    protected IAccumulatorLengthFunction<E> accumulatorLengthFunction;
+
+    public AbstractAccumulator(double byteSizeLimit){
+        this.byteSizeLimit = byteSizeLimit;
+        this.accumulatorLengthFunction = accumulatorLengthFunction();
+    }
+
+
+    @Override
+    public boolean add(E obj) {
+
+        double size = objectSize(obj);
+
+        if(shouldBecomeFull(size)){
+            return false;
+        }
+
+        accumulator.add(obj);
+        accumulatorSize.addAndGet(size);
+        return true;
+    }
+
+
+    private double objectSize(E obj)
+    {
+        return accumulatorLengthFunction.apply(obj);
+    }
+
+    private boolean shouldBecomeFull(double size){
+        return (accumulatorSize.get() + size > byteSizeLimit);
+    }
+}