You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@wayang.apache.org by be...@apache.org on 2021/04/06 16:18:03 UTC
[incubator-wayang] branch debugger updated (4c01bde -> a3342a0)
This is an automated email from the ASF dual-hosted git repository.
bertty pushed a change to branch debugger
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git.
from 4c01bde [WAYANG-28] adding missing license header's
new 093d25c [WAYANG-28] Add sniffer components templates
new 0efab7b [WAYANG-28] Add Shipper components templates
new d4dcae5 [WAYANG-28] Rename Sniffer.java to Sniff.java
new f02962d [WAYANG-28] Add basic logic to HackitSniffer
new 2afe801 [WAYANG-28] Add BasicCloner
new d227907 [WAYANG-28] add EmptyInjector implementation
new a3342a0 [WAYANG-28] add OneElementIterator.java
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../wayang-hackit/wayang-hackit-core/pom.xml | 12 ++-
.../hackit/core/iterator/ElementPlusIterator.java | 36 +++++++
.../hackit/core/iterator/OneElementIterator.java | 58 +++++++++++
.../plugin/hackit/core/sniffer/HackitSniffer.java | 115 +++++++++++++++++++++
.../plugin/hackit/core/sniffer/actor/Actor.java | 9 +-
.../hackit/core/sniffer/clone/BasicCloner.java | 12 +++
.../plugin/hackit/core/sniffer/clone/Cloner.java | 9 +-
.../hackit/core/sniffer/inject/EmptyInjector.java | 49 +++++++++
.../inject/Injector.java} | 15 ++-
.../hackit/core/sniffer/shipper/PSProtocol.java | 11 +-
.../hackit/core/sniffer/shipper/Shipper.java | 108 +++++++++++++++++++
.../shipper/receiver/BufferReceiver.java} | 48 ++++-----
.../shipper/receiver/Receiver.java} | 17 +--
.../hackit/core/sniffer/shipper/sender/Sender.java | 11 +-
.../plugin/hackit/core/sniffer/sniff/Sniff.java | 9 +-
15 files changed, 444 insertions(+), 75 deletions(-)
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/ElementPlusIterator.java
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/OneElementIterator.java
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSniffer.java
copy wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/PythonAPI.java => wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/actor/Actor.java (82%)
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/clone/BasicCloner.java
copy wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/PythonAPI.java => wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/clone/Cloner.java (82%)
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/EmptyInjector.java
copy wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/{tuple/header/HeaderBuilder.java => sniffer/inject/Injector.java} (71%)
copy wayang-platforms/wayang-spark/src/main/java/org/apache/wayang/spark/channels/FileChannels.java => wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/PSProtocol.java (78%)
create mode 100644 wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
copy wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/{tags/DebugTag.java => sniffer/shipper/receiver/BufferReceiver.java} (55%)
copy wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/{iterator/HackitIterator.java => sniffer/shipper/receiver/Receiver.java} (70%)
copy wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/PythonAPI.java => wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/sender/Sender.java (78%)
copy wayang-api/wayang-api-python/src/main/java/org/apache/wayang/api/python/PythonAPI.java => wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniff.java (82%)
[incubator-wayang] 02/07: [WAYANG-28] Add Shipper components
templates
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bertty pushed a commit to branch debugger
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 0efab7baa550b3d8928f335fb45deb9211a342b1
Author: Bertty Contreras-Rojas <be...@scalytics.io>
AuthorDate: Tue Apr 6 11:21:27 2021 -0400
[WAYANG-28] Add Shipper components templates
---
.../shipper/{Shipper.java => PSProtocol.java} | 7 +-
.../hackit/core/sniffer/shipper/Shipper.java | 89 +++++++++++++++++++++-
.../{Shipper.java => receiver/BufferReceiver.java} | 31 +++++++-
.../{Shipper.java => receiver/Receiver.java} | 15 +++-
.../shipper/{Shipper.java => sender/Sender.java} | 12 ++-
5 files changed, 146 insertions(+), 8 deletions(-)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/PSProtocol.java
similarity index 83%
copy from wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
copy to wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/PSProtocol.java
index 967f693..6a19794 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/PSProtocol.java
@@ -17,5 +17,10 @@
*/
package org.apache.wayang.plugin.hackit.core.sniffer.shipper;
-public interface Shipper {
+/**
+ * Publish and Subscribed Protocol
+ */
+public interface PSProtocol {
+ public PSProtocol addTopic(String... topic);
+ public PSProtocol addExchange(String exchange);
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
index 967f693..969d8e8 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
@@ -17,5 +17,92 @@
*/
package org.apache.wayang.plugin.hackit.core.sniffer.shipper;
-public interface Shipper {
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.receiver.Receiver;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.sender.Sender;
+
+import java.io.Serializable;
+import java.util.Iterator;
+
+public abstract class Shipper<T, ST, SenderObj extends Sender<ST>, ReceiverObj extends Receiver<T>> implements Iterator<T>, Serializable {
+
+ protected Sender sender_instance;
+ protected Receiver receiver_instance;
+
+ protected abstract Sender createSenderInstance();
+ protected abstract Receiver createReceiverInstance();
+
+ /** Connect with the a Message queue service*/
+ public void publish(ST value){
+ if(this.sender_instance == null){
+ throw new RuntimeException("The Sender of the Shipper is not instanciated");
+ }
+ this.sender_instance.send(value);
+ }
+
+ /** To subscribe as a producer */
+ public void subscribeAsProducer(){
+ this.sender_instance = this.createSenderInstance();
+ this.sender_instance.init();
+ }
+
+ public void subscribeAsProducer(String... topic){
+ this.subscribeAsProducer("default", topic);
+ }
+
+ public void subscribeAsProducer(String metatopic, String... topic){
+ this.subscribeAsProducer();
+ ((PSProtocol)this.sender_instance)
+ .addExchange(metatopic)
+ .addTopic(topic)
+ ;
+ }
+
+ /** Close connection */
+ public void unsubscribeAsProducer(){
+ if( this.sender_instance == null) return;
+ this.sender_instance.close();
+ }
+
+ /** To subscribe/unsubscribe as a consumer
+ * metatopic correspond to EXCHANGE_NAME
+ * topics correspond to bindingKeys
+ * */
+ public void subscribeAsConsumer(){
+ this.receiver_instance = this.createReceiverInstance();
+ this.receiver_instance.init();
+ }
+ public void subscribeAsConsumer(String... topic){
+ this.subscribeAsProducer("default", topic);
+ }
+ public void subscribeAsConsumer(String metatopic, String... topic){
+ this.subscribeAsConsumer();
+ ((PSProtocol)this.receiver_instance)
+ .addExchange(metatopic)
+ .addTopic(topic)
+ ;
+ }
+
+ /** Close connection */
+ public void unsubscribeAsConsumer() {
+ if( this.receiver_instance == null) return;
+ this.receiver_instance.close();
+ }
+
+ public void close(){
+ this.unsubscribeAsConsumer();
+ this.unsubscribeAsProducer();
+ }
+
+ @Override
+ public abstract boolean hasNext();
+
+ @Override
+ public abstract T next();
+
+ public Iterator<T> getNexts(){
+ if( this.receiver_instance == null){
+ throw new RuntimeException("The Receiver of the Shipper is not instanciated");
+ }
+ return this.receiver_instance.getElements();
+ }
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/receiver/BufferReceiver.java
similarity index 51%
copy from wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
copy to wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/receiver/BufferReceiver.java
index 967f693..0a93dae 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/receiver/BufferReceiver.java
@@ -15,7 +15,34 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.wayang.plugin.hackit.core.sniffer.shipper;
+package org.apache.wayang.plugin.hackit.core.sniffer.shipper.receiver;
-public interface Shipper {
+import java.io.Serializable;
+import java.util.Queue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+public class BufferReceiver<T> implements Serializable {
+ //TODO implement the doble buffering
+ private transient Queue<T> queue;
+
+ //TODO implement the server to receive the messages
+ public boolean start(){
+ return true;
+ }
+
+ //TODO registrer on the rest of the worker
+ public boolean register(){
+ return true;
+ }
+
+ public boolean existQueue(){
+ return false;
+ }
+
+ public void put(T value){
+ if(this.queue == null){
+ this.queue = new LinkedBlockingQueue<>();
+ }
+ this.queue.add(value);
+ }
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/receiver/Receiver.java
similarity index 69%
copy from wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
copy to wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/receiver/Receiver.java
index 967f693..dd120c0 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/receiver/Receiver.java
@@ -15,7 +15,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.wayang.plugin.hackit.core.sniffer.shipper;
+package org.apache.wayang.plugin.hackit.core.sniffer.shipper.receiver;
-public interface Shipper {
+import java.io.Serializable;
+import java.util.Iterator;
+
+public abstract class Receiver<T> implements Serializable {
+
+ private transient BufferReceiver<T> bufferReceive;
+
+ public abstract void init();
+
+ public abstract Iterator<T> getElements();
+
+ public abstract void close();
}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/sender/Sender.java
similarity index 77%
copy from wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
copy to wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/sender/Sender.java
index 967f693..89eb57d 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/sender/Sender.java
@@ -15,7 +15,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.wayang.plugin.hackit.core.sniffer.shipper;
+package org.apache.wayang.plugin.hackit.core.sniffer.shipper.sender;
-public interface Shipper {
+import java.io.Serializable;
+
+public interface Sender<T> extends Serializable {
+
+ public void init();
+
+ public void send(T value);
+
+ public void close();
}
[incubator-wayang] 06/07: [WAYANG-28] add EmptyInjector
implementation
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bertty pushed a commit to branch debugger
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit d22790782f5623c2fdb29398a8c6f6aa16d9424a
Author: Bertty Contreras-Rojas <be...@scalytics.io>
AuthorDate: Tue Apr 6 11:53:17 2021 -0400
[WAYANG-28] add EmptyInjector implementation
---
.../hackit/core/iterator/ElementPlusIterator.java | 36 ++++++++++++++++
.../hackit/core/sniffer/inject/EmptyInjector.java | 49 ++++++++++++++++++++++
2 files changed, 85 insertions(+)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/ElementPlusIterator.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/ElementPlusIterator.java
new file mode 100644
index 0000000..73353bc
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/ElementPlusIterator.java
@@ -0,0 +1,36 @@
+package org.apache.wayang.plugin.hackit.core.iterator;
+
+import java.util.Iterator;
+
+public class ElementPlusIterator<T> implements Iterator<T> {
+
+ private boolean element_consumed = false;
+ private T element;
+ private Iterator<T> iterator;
+
+
+ public ElementPlusIterator(T element, Iterator<T> iterator) {
+ if(element == null){
+ throw new RuntimeException("the element can't be null");
+ }
+ this.element = element;
+ this.iterator = iterator;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if( ! this.element_consumed ){
+ return true;
+ }
+ return this.iterator.hasNext();
+ }
+
+ @Override
+ public T next() {
+ if( ! this.element_consumed ){
+ this.element_consumed = true;
+ return this.element;
+ }
+ return this.iterator.next();
+ }
+}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/EmptyInjector.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/EmptyInjector.java
new file mode 100644
index 0000000..689882d
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/EmptyInjector.java
@@ -0,0 +1,49 @@
+package org.apache.wayang.plugin.hackit.core.sniffer.inject;
+
+import org.apache.wayang.plugin.hackit.core.iterator.ElementPlusIterator;
+
+import java.util.Iterator;
+import java.util.Spliterator;
+import java.util.Spliterators;
+import java.util.stream.StreamSupport;
+
+public class EmptyInjector<T> implements Injector<T>{
+
+ @Override
+ public Iterator<T> inject(T element, Iterator<T> iterator) {
+ return
+ StreamSupport.stream(
+ Spliterators.spliteratorUnknownSize(
+ new ElementPlusIterator<T>(
+ element,
+ iterator
+ ),
+ Spliterator.ORDERED
+ ),
+ false
+ ).map(
+ (T _element) -> {
+ if (this.is_halt_job(_element)) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ return _element;
+ }
+ ).filter(
+ (T _element) -> ! this.is_skip_element(_element)
+ ).iterator();
+ }
+
+ @Override
+ public boolean is_skip_element(T element) {
+ return false;
+ }
+
+ @Override
+ public boolean is_halt_job(T element) {
+ return false;
+ }
+}
[incubator-wayang] 07/07: [WAYANG-28] add OneElementIterator.java
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bertty pushed a commit to branch debugger
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit a3342a09368aea8d2878f11c88e78db6a38115d7
Author: Bertty Contreras-Rojas <be...@scalytics.io>
AuthorDate: Tue Apr 6 12:07:23 2021 -0400
[WAYANG-28] add OneElementIterator.java
---
.../hackit/core/iterator/OneElementIterator.java | 58 ++++++++++++++++++++++
1 file changed, 58 insertions(+)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/OneElementIterator.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/OneElementIterator.java
new file mode 100644
index 0000000..5f5c0dd
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/iterator/OneElementIterator.java
@@ -0,0 +1,58 @@
+package org.apache.wayang.plugin.hackit.core.iterator;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+public class OneElementIterator<T> implements Iterator<T>, Iterable<T>{
+
+ private final boolean removeAllowed;
+ private boolean beforeFirst;
+ private boolean removed;
+ private T object;
+
+ public OneElementIterator(T object) {
+ this(object, true);
+ }
+
+ public OneElementIterator(T object, boolean removeAllowed) {
+ this.beforeFirst = true;
+ this.removed = false;
+ this.object = object;
+ this.removeAllowed = removeAllowed;
+ }
+
+ public boolean hasNext() {
+ return this.beforeFirst && !this.removed;
+ }
+
+ public T next() {
+ if (this.beforeFirst && !this.removed) {
+ this.beforeFirst = false;
+ return this.object;
+ } else {
+ throw new NoSuchElementException();
+ }
+ }
+
+ public void remove() {
+ if (this.removeAllowed) {
+ if (!this.removed && !this.beforeFirst) {
+ this.object = null;
+ this.removed = true;
+ } else {
+ throw new IllegalStateException();
+ }
+ } else {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ public void reset() {
+ this.beforeFirst = true;
+ }
+
+ @Override
+ public Iterator<T> iterator() {
+ return this;
+ }
+}
[incubator-wayang] 04/07: [WAYANG-28] Add basic logic to
HackitSniffer
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bertty pushed a commit to branch debugger
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit f02962d518f88e904a185b7fe69d09a37054a7ad
Author: Bertty Contreras-Rojas <be...@scalytics.io>
AuthorDate: Tue Apr 6 11:23:01 2021 -0400
[WAYANG-28] Add basic logic to HackitSniffer
---
.../plugin/hackit/core/sniffer/HackitSniffer.java | 96 +++++++++++++++++++++-
1 file changed, 95 insertions(+), 1 deletion(-)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSniffer.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSniffer.java
index 9e6c77e..02342ff 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSniffer.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSniffer.java
@@ -17,5 +17,99 @@
*/
package org.apache.wayang.plugin.hackit.core.sniffer;
-public class HackitSniffer {
+import org.apache.wayang.plugin.hackit.core.sniffer.actor.Actor;
+import org.apache.wayang.plugin.hackit.core.sniffer.clone.Cloner;
+import org.apache.wayang.plugin.hackit.core.sniffer.inject.Injector;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.Shipper;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.sender.Sender;
+import org.apache.wayang.plugin.hackit.core.sniffer.shipper.receiver.Receiver;
+import org.apache.wayang.plugin.hackit.core.sniffer.sniff.Sniff;
+import org.apache.wayang.plugin.hackit.core.tuple.HackitTuple;
+
+import java.io.Serializable;
+import java.util.Iterator;
+import java.util.function.Function;
+
+public class HackitSniffer<K, T, SentType, SenderObj extends Sender<SentType>, ReceiverObj extends Receiver<HackitTuple<K,T>> > implements Function<HackitTuple<K, T>, Iterator<HackitTuple<K, T>>>, Serializable {
+
+ private transient boolean not_first = false;
+ private Injector<HackitTuple<K, T>> hackItInjector;
+
+ private Actor<HackitTuple<K, T>> actorFunction;
+
+ private Shipper<HackitTuple<K, T>, SentType, SenderObj, ReceiverObj> shipper;
+
+ private Sniff<HackitTuple<K, T>> hackItSniff;
+ private Cloner<HackitTuple<K, T>, SentType> hackItCloner;
+
+ public HackitSniffer(Injector<HackitTuple<K, T>> hackItInjector, Actor<HackitTuple<K, T>> actorFunction, Shipper<HackitTuple<K, T>, SentType, SenderObj, ReceiverObj> shipper, Sniff<HackitTuple<K, T>> hackItSniff, Cloner<HackitTuple<K, T>, SentType> hackItCloner) {
+ this.hackItInjector = hackItInjector;
+ this.actorFunction = actorFunction;
+ this.shipper = shipper;
+ this.hackItSniff = hackItSniff;
+ this.hackItCloner = hackItCloner;
+ this.not_first = false;
+ }
+
+ public HackitSniffer() {
+ //TODO this over configuration file
+ this.not_first = false;
+ }
+
+ @Override
+ public Iterator<HackitTuple<K, T>> apply(HackitTuple<K, T> ktHackItTuple) {
+ if(!this.not_first){
+ this.shipper.subscribeAsProducer();
+ this.shipper.subscribeAsConsumer();
+ this.not_first = true;
+ }
+
+ if(this.hackItSniff.sniff(ktHackItTuple)){
+ if(this.actorFunction.is_sendout(ktHackItTuple)){
+ this.shipper.publish(
+ this.hackItCloner.clone(ktHackItTuple)
+ );
+ }
+ }
+ Iterator<HackitTuple<K, T>> inyection = this.shipper.getNexts();
+
+ return this.hackItInjector.inject(ktHackItTuple, inyection);
+ }
+
+ public HackitSniffer<K, T, SentType, SenderObj, ReceiverObj> setHackItInjector(Injector<HackitTuple<K, T>> hackItInjector) {
+ this.hackItInjector = hackItInjector;
+ return this;
+ }
+
+ public HackitSniffer<K, T, SentType, SenderObj, ReceiverObj> setActorFunction(Actor<HackitTuple<K, T>> actorFunction) {
+ this.actorFunction = actorFunction;
+ return this;
+ }
+
+ public HackitSniffer<K, T, SentType, SenderObj, ReceiverObj> setShipper(Shipper<HackitTuple<K, T>, SentType, SenderObj, ReceiverObj> shipper) {
+ this.shipper = shipper;
+ return this;
+ }
+
+ public HackitSniffer<K, T, SentType, SenderObj, ReceiverObj> setHackItSniff(Sniff<HackitTuple<K, T>> hackItSniff) {
+ this.hackItSniff = hackItSniff;
+ return this;
+ }
+
+ public HackitSniffer<K, T, SentType, SenderObj, ReceiverObj> setHackItCloner(Cloner<HackitTuple<K, T>, SentType> hackItCloner) {
+ this.hackItCloner = hackItCloner;
+ return this;
+ }
+
+ @Override
+ public String toString() {
+ return "HackItSniffer{" +
+ "\nfirst=" + not_first +
+ ",\n hackItInjector=" + hackItInjector +
+ ",\n actorFunction=" + actorFunction +
+ ",\n shipper=" + shipper +
+ ",\n hackItSniff=" + hackItSniff +
+ ",\n hackItCloner=" + hackItCloner +
+ "\n}";
+ }
}
[incubator-wayang] 03/07: [WAYANG-28] Rename Sniffer.java to
Sniff.java
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bertty pushed a commit to branch debugger
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit d4dcae55056d94c19da25f04f1de3fcebf458332
Author: Bertty Contreras-Rojas <be...@scalytics.io>
AuthorDate: Tue Apr 6 11:21:54 2021 -0400
[WAYANG-28] Rename Sniffer.java to Sniff.java
---
.../plugin/hackit/core/sniffer/sniff/{Sniffer.java => Sniff.java} | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniffer.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniff.java
similarity index 94%
rename from wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniffer.java
rename to wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniff.java
index 9c2fc65..44bfbbb 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniffer.java
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniff.java
@@ -19,7 +19,7 @@ package org.apache.wayang.plugin.hackit.core.sniffer.sniff;
import java.io.Serializable;
-public interface Sniffer<I> extends Serializable {
+public interface Sniff<I> extends Serializable {
public boolean sniff(I input);
}
[incubator-wayang] 01/07: [WAYANG-28] Add sniffer components
templates
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bertty pushed a commit to branch debugger
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 093d25cfe6f949709424385066223751848947c3
Author: Bertty Contreras-Rojas <be...@scalytics.io>
AuthorDate: Tue Apr 6 11:02:21 2021 -0400
[WAYANG-28] Add sniffer components templates
---
.../plugin/hackit/core/sniffer/HackitSniffer.java | 21 +++++++++++++++
.../plugin/hackit/core/sniffer/actor/Actor.java | 25 ++++++++++++++++++
.../plugin/hackit/core/sniffer/clone/Cloner.java | 25 ++++++++++++++++++
.../hackit/core/sniffer/inject/Injector.java | 30 ++++++++++++++++++++++
.../hackit/core/sniffer/shipper/Shipper.java | 21 +++++++++++++++
.../plugin/hackit/core/sniffer/sniff/Sniffer.java | 25 ++++++++++++++++++
6 files changed, 147 insertions(+)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSniffer.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSniffer.java
new file mode 100644
index 0000000..9e6c77e
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/HackitSniffer.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.wayang.plugin.hackit.core.sniffer;
+
+public class HackitSniffer {
+}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/actor/Actor.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/actor/Actor.java
new file mode 100644
index 0000000..b49a7c4
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/actor/Actor.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.wayang.plugin.hackit.core.sniffer.actor;
+
+import java.io.Serializable;
+
+public interface Actor<T> extends Serializable {
+
+ public boolean is_sendout(T value);
+}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/clone/Cloner.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/clone/Cloner.java
new file mode 100644
index 0000000..ccf99ab
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/clone/Cloner.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.wayang.plugin.hackit.core.sniffer.clone;
+
+import java.io.Serializable;
+
+public interface Cloner<I, O> extends Serializable {
+
+ public O clone(I input);
+}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/Injector.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/Injector.java
new file mode 100644
index 0000000..3ca4563
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/inject/Injector.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.wayang.plugin.hackit.core.sniffer.inject;
+
+import java.io.Serializable;
+import java.util.Iterator;
+
+public interface Injector<T> extends Serializable {
+
+ public Iterator<T> inject(T element, Iterator<T> iterator);
+
+ public boolean is_skip_element(T element);
+
+ public boolean is_halt_job(T element);
+}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
new file mode 100644
index 0000000..967f693
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/shipper/Shipper.java
@@ -0,0 +1,21 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.wayang.plugin.hackit.core.sniffer.shipper;
+
+public interface Shipper {
+}
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniffer.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniffer.java
new file mode 100644
index 0000000..9c2fc65
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/sniff/Sniffer.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.wayang.plugin.hackit.core.sniffer.sniff;
+
+import java.io.Serializable;
+
+public interface Sniffer<I> extends Serializable {
+
+ public boolean sniff(I input);
+}
[incubator-wayang] 05/07: [WAYANG-28] Add BasicCloner
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
bertty pushed a commit to branch debugger
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 2afe80181a5c11f6abe5c4d7742e221a2d80fc04
Author: Bertty Contreras-Rojas <be...@scalytics.io>
AuthorDate: Tue Apr 6 11:47:45 2021 -0400
[WAYANG-28] Add BasicCloner
---
wayang-plugins/wayang-hackit/wayang-hackit-core/pom.xml | 12 ++++++++----
.../wayang/plugin/hackit/core/sniffer/clone/BasicCloner.java | 12 ++++++++++++
2 files changed, 20 insertions(+), 4 deletions(-)
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/pom.xml b/wayang-plugins/wayang-hackit/wayang-hackit-core/pom.xml
index eb947ba..0f1fe09 100644
--- a/wayang-plugins/wayang-hackit/wayang-hackit-core/pom.xml
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/pom.xml
@@ -29,9 +29,13 @@
<artifactId>wayang-hackit-core</artifactId>
- <properties>
- <maven.compiler.source>8</maven.compiler.source>
- <maven.compiler.target>8</maven.compiler.target>
- </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-lang3</artifactId>
+ <version>3.12.0</version>
+ </dependency>
+ </dependencies>
</project>
diff --git a/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/clone/BasicCloner.java b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/clone/BasicCloner.java
new file mode 100644
index 0000000..47970a6
--- /dev/null
+++ b/wayang-plugins/wayang-hackit/wayang-hackit-core/src/main/java/org/apache/wayang/plugin/hackit/core/sniffer/clone/BasicCloner.java
@@ -0,0 +1,12 @@
+package org.apache.wayang.plugin.hackit.core.sniffer.clone;
+
+import org.apache.commons.lang3.SerializationUtils;
+
+import java.io.Serializable;
+
+public class BasicCloner<I> implements Cloner<I, byte[]> {
+ @Override
+ public byte[] clone(I input) {
+ return SerializationUtils.serialize((Serializable) input);
+ }
+}