streams-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sblack...@apache.org
Subject [01/14] git commit: adding push provider, reverting change to streams provider
Date Fri, 08 Aug 2014 20:10:16 GMT
Repository: incubator-streams
Updated Branches:
  refs/heads/master 41a80ed2e -> d309a5987


adding push provider, reverting change to streams provider


Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/d9e0a384
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/d9e0a384
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/d9e0a384

Branch: refs/heads/master
Commit: d9e0a384cda3820d53d40d47329fd65fb09c1282
Parents: f9d484b
Author: Steve Blackmon <sblackmon@w2odigital.com>
Authored: Sun Jul 27 19:13:25 2014 -0500
Committer: Steve Blackmon <sblackmon@w2odigital.com>
Committed: Sun Jul 27 19:14:12 2014 -0500

----------------------------------------------------------------------
 .../datasift/provider/DatasiftPushProvider.java | 154 +++++++++++++++++++
 .../com/datasift/DatasiftPushConfiguration.json |  17 ++
 .../datasift/DatasiftStreamConfiguration.json   |  17 ++
 3 files changed, 188 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d9e0a384/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftPushProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftPushProvider.java
b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftPushProvider.java
new file mode 100644
index 0000000..196f504
--- /dev/null
+++ b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftPushProvider.java
@@ -0,0 +1,154 @@
+/*
+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.streams.datasift.provider;
+
+import com.datasift.client.DataSiftClient;
+import com.datasift.client.stream.DeletedInteraction;
+import com.datasift.client.stream.Interaction;
+import com.datasift.client.stream.StreamEventListener;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Queues;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProvider;
+import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.datasift.DatasiftConfiguration;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.util.Map;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Requires Java Version 1.7!
+ * {@code DatasiftStreamProvider} is an implementation of the {@link org.apache.streams.core.StreamsProvider}
interface.  The provider
+ * uses the Datasift java api to make connections. A single provider creates one connection
per StreamHash in the configuration.
+ */
+public class DatasiftPushProvider implements StreamsProvider {
+
+    private final static Logger LOGGER = LoggerFactory.getLogger(DatasiftPushProvider.class);
+
+    private DatasiftConfiguration config;
+    protected ConcurrentLinkedQueue<Interaction> interactions = new ConcurrentLinkedQueue<Interaction>();
+    private Map<String, DataSiftClient> clients;
+    private StreamEventListener eventListener;
+    private ObjectMapper mapper;
+
+    public DatasiftPushProvider() {
+
+    }
+
+    // to set up a webhook we need to be able to return a reference to this queue
+    public Queue<Interaction> getInteractions() {
+        return interactions;
+    }
+
+    @Override
+    public void startStream() {
+
+        Preconditions.checkNotNull(this.config);
+        Preconditions.checkNotNull(this.config.getApiKey());
+        Preconditions.checkNotNull(this.config.getUserName());
+
+    }
+
+    /**
+     * Shuts down all open streams from datasift.
+     */
+    public void stop() {
+    }
+
+    // PRIME EXAMPLE OF WHY WE NEED NEW INTERFACES FOR PROVIDERS
+    @Override
+    //This is a hack.  It is only like this because of how perpetual streams work at the
moment.  Read list server to debate/vote for new interfaces.
+    public StreamsResultSet readCurrent() {
+        Queue<StreamsDatum> datums = Queues.newConcurrentLinkedQueue();
+        StreamsDatum datum = null;
+        Interaction interaction;
+        while (!this.interactions.isEmpty()) {
+            interaction = this.interactions.poll();
+            try {
+                datum = new StreamsDatum(this.mapper.writeValueAsString(interaction.getData()),
interaction.getData().get("interaction").get("id").textValue());
+            } catch (JsonProcessingException jpe) {
+                LOGGER.error("Exception while converting Interaction to String : {}", jpe);
+            }
+            if (datum != null) {
+                while (!datums.offer(datum)) {
+                    Thread.yield();
+                }
+            }
+
+        }
+        return new StreamsResultSet(datums);
+    }
+
+    @Override
+    public StreamsResultSet readNew(BigInteger sequence) {
+        return null;
+    }
+
+    public StreamsResultSet readRange(DateTime start, DateTime end) {
+        return null;
+    }
+
+    @Override
+    public boolean isRunning() {
+        return this.clients != null && this.clients.size() > 0;
+    }
+
+    @Override
+    public void prepare(Object configurationObject) {
+        this.interactions = new ConcurrentLinkedQueue<Interaction>();
+        this.clients = Maps.newHashMap();
+        this.mapper = StreamsJacksonMapper.getInstance();
+    }
+
+    @Override
+    public void cleanUp() {
+        stop();
+    }
+
+    public DatasiftConfiguration getConfig() {
+        return config;
+    }
+
+    public void setConfig(DatasiftConfiguration config) {
+        this.config = config;
+    }
+
+
+    /**
+     * THIS CLASS NEEDS TO BE REPLACED/OVERRIDDEN BY ALL USERS. TWITTERS TERMS OF SERVICE
SAYS THAT EVERYONE MUST
+     * DELETE TWEETS FROM THEIR DATA STORE IF THEY RECEIVE A DELETE NOTICE.
+     */
+    public static class DeleteHandler extends StreamEventListener {
+
+        public void onDelete(DeletedInteraction di) {
+            //go off and delete the interaction if you have it stored. This is a strict requirement!
+            LOGGER.info("DELETED:\n " + di);
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d9e0a384/streams-contrib/streams-provider-datasift/src/main/jsonschema/com/datasift/DatasiftPushConfiguration.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/jsonschema/com/datasift/DatasiftPushConfiguration.json
b/streams-contrib/streams-provider-datasift/src/main/jsonschema/com/datasift/DatasiftPushConfiguration.json
new file mode 100644
index 0000000..bb65ef0
--- /dev/null
+++ b/streams-contrib/streams-provider-datasift/src/main/jsonschema/com/datasift/DatasiftPushConfiguration.json
@@ -0,0 +1,17 @@
+{
+    "type": "object",
+    "$schema": "http://json-schema.org/draft-03/schema",
+    "id": "#",
+    "javaType" : "org.apache.streams.datasift.DatasiftPushConfiguration",
+    "extends": {"$ref":"DatasiftConfiguration.json"},
+    "javaInterfaces": ["java.io.Serializable"],
+    "properties": {
+        "streamHash": {
+            "type": "array",
+            "minItems": 1,
+            "items": {
+                "type": "string"
+            }
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/d9e0a384/streams-contrib/streams-provider-datasift/src/main/jsonschema/com/datasift/DatasiftStreamConfiguration.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/jsonschema/com/datasift/DatasiftStreamConfiguration.json
b/streams-contrib/streams-provider-datasift/src/main/jsonschema/com/datasift/DatasiftStreamConfiguration.json
new file mode 100644
index 0000000..91a9974
--- /dev/null
+++ b/streams-contrib/streams-provider-datasift/src/main/jsonschema/com/datasift/DatasiftStreamConfiguration.json
@@ -0,0 +1,17 @@
+{
+    "type": "object",
+    "$schema": "http://json-schema.org/draft-03/schema",
+    "id": "#",
+    "javaType" : "org.apache.streams.datasift.DatasiftStreamConfiguration",
+    "extends": {"$ref":"DatasiftConfiguration.json"},
+    "javaInterfaces": ["java.io.Serializable"],
+    "properties": {
+        "streamHash": {
+            "type": "array",
+            "minItems": 1,
+            "items": {
+                "type": "string"
+            }
+        }
+    }
+}
\ No newline at end of file


Mime
View raw message