You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2023/01/11 14:28:12 UTC
[streampipes] branch SP-1077 created (now 71e5bca63)
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a change to branch SP-1077
in repository https://gitbox.apache.org/repos/asf/streampipes.git
at 71e5bca63 [#1077] Remove legacy method getNElements
This branch includes the following new commits:
new 71e5bca63 [#1077] Remove legacy method getNElements
The 1 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.
[streampipes] 01/01: [#1077] Remove legacy method getNElements
Posted by ze...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch SP-1077
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 71e5bca63a17578a2b4d23930fd6a4f55bc146fb
Author: Philipp Zehnder <te...@users.noreply.github.com>
AuthorDate: Wed Jan 11 15:27:40 2023 +0100
[#1077] Remove legacy method getNElements
---
.../extensions/api/connect/IProtocol.java | 5 ----
.../connect/iiot/protocol/set/FileProtocol.java | 28 ------------------
.../connect/iiot/protocol/set/HttpProtocol.java | 29 ------------------
.../iiot/protocol/stream/BrokerProtocol.java | 13 ---------
.../iiot/protocol/stream/FileStreamProtocol.java | 27 -----------------
.../iiot/protocol/stream/HttpServerProtocol.java | 7 -----
.../iiot/protocol/stream/HttpStreamProtocol.java | 34 ----------------------
7 files changed, 143 deletions(-)
diff --git a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IProtocol.java b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IProtocol.java
index 6672b4668..2d9982536 100644
--- a/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IProtocol.java
+++ b/streampipes-extensions-api/src/main/java/org/apache/streampipes/extensions/api/connect/IProtocol.java
@@ -23,9 +23,6 @@ import org.apache.streampipes.model.connect.grounding.ProtocolDescription;
import org.apache.streampipes.model.connect.guess.GuessSchema;
import org.apache.streampipes.model.schema.EventSchema;
-import java.util.List;
-import java.util.Map;
-
public interface IProtocol extends Connector {
IProtocol getInstance(ProtocolDescription protocolDescription,
@@ -36,8 +33,6 @@ public interface IProtocol extends Connector {
GuessSchema getGuessSchema() throws ParseException;
- List<Map<String, Object>> getNElements(int n) throws ParseException;
-
void run(IAdapterPipeline adapterPipeline) throws AdapterException;
/*
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/set/FileProtocol.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/set/FileProtocol.java
index ff852297f..347826e12 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/set/FileProtocol.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/set/FileProtocol.java
@@ -45,9 +45,7 @@ import org.slf4j.LoggerFactory;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.InputStream;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
public class FileProtocol extends Protocol {
@@ -134,32 +132,6 @@ public class FileProtocol extends Protocol {
}
- @Override
- public List<Map<String, Object>> getNElements(int n) throws ParseException {
- List<Map<String, Object>> result = new ArrayList<>();
-
- List<byte[]> dataByteArray = new ArrayList<>();
- try {
- InputStream dataInputStream = FileProtocolUtils.getFileInputStream(this.selectedFilename);
- dataByteArray = parser.parseNEvents(dataInputStream, n);
- } catch (FileNotFoundException e) {
- e.printStackTrace();
- }
-
- // Check that result size is n. Currently just an error is logged. Maybe change to an exception
- if (dataByteArray.size() < n) {
- logger.error("Error in File Protocol! User required: " + n + " elements but the resource just had: "
- + dataByteArray.size());
- }
-
- for (byte[] b : dataByteArray) {
- result.add(format.parse(b));
- }
-
- return result;
- }
-
-
@Override
public String getId() {
return ID;
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/set/HttpProtocol.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/set/HttpProtocol.java
index c73c06b19..d72001c11 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/set/HttpProtocol.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/set/HttpProtocol.java
@@ -42,9 +42,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.InputStream;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
public class HttpProtocol extends Protocol {
@@ -122,27 +120,6 @@ public class HttpProtocol extends Protocol {
return result;
}
- @Override
- public List<Map<String, Object>> getNElements(int n) throws ParseException {
-
- List<Map<String, Object>> result = new ArrayList<>();
-
- InputStream dataInputStream = getDataFromEndpoint();
-
- List<byte[]> dataByteArray = parser.parseNEvents(dataInputStream, n);
-
- // Check that result size is n. Currently just an error is logged. Maybe change to an exception
- if (dataByteArray.size() < n) {
- logger.error("Error in HttpProtocol! User required: " + n + " elements but the resource just had: "
- + dataByteArray.size());
- }
-
- for (byte[] b : dataByteArray) {
- result.add(format.parse(b));
- }
-
- return result;
- }
public InputStream getDataFromEndpoint() throws ParseException {
InputStream result = null;
@@ -153,12 +130,6 @@ public class HttpProtocol extends Protocol {
.socketTimeout(100000)
.execute().returnContent().asStream();
-// if (s.startsWith("ï")) {
-// s = s.substring(3);
-// }
-
-// result = IOUtils.toInputStream(s, "UTF-8");
-
} catch (IOException e) {
throw new ParseException("Could not receive Data from: " + url);
}
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/BrokerProtocol.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/BrokerProtocol.java
index 26ec15241..5885b3fd8 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/BrokerProtocol.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/BrokerProtocol.java
@@ -24,9 +24,7 @@ import org.apache.streampipes.extensions.management.connect.adapter.guess.Schema
import org.apache.streampipes.extensions.management.connect.adapter.model.generic.Protocol;
import org.apache.streampipes.model.connect.guess.GuessSchema;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
public abstract class BrokerProtocol extends Protocol {
@@ -55,17 +53,6 @@ public abstract class BrokerProtocol extends Protocol {
}
}
- @Override
- public List<Map<String, Object>> getNElements(int n) throws ParseException {
- List<byte[]> resultEventsByte = getNByteElements(n);
- List<Map<String, Object>> result = new ArrayList<>();
- for (byte[] event : resultEventsByte) {
- result.add(format.parse(event));
- }
-
- return result;
- }
-
protected abstract List<byte[]> getNByteElements(int n) throws ParseException;
}
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileStreamProtocol.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileStreamProtocol.java
index 280ca40cf..d47a0f4b4 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileStreamProtocol.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/FileStreamProtocol.java
@@ -55,7 +55,6 @@ import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
-import java.util.Map;
public class FileStreamProtocol extends Protocol {
@@ -63,9 +62,7 @@ public class FileStreamProtocol extends Protocol {
public static final String ID = "org.apache.streampipes.connect.iiot.protocol.stream.file";
- //private String filePath;
private String selectedFileName;
- // private String timestampKey;
private boolean replaceTimestamp;
private float speedUp;
private int timeBetweenReplay;
@@ -165,7 +162,6 @@ public class FileStreamProtocol extends Protocol {
StaticPropertyExtractor.from(protocolDescription.getConfig(), new ArrayList<>());
List<String> replaceTimestampStringList = extractor.selectedMultiValues("replaceTimestamp", String.class);
-// String replaceTimestampString = extractor.selectedSingleValueOption("replaceTimestamp");
boolean replaceTimestamp = replaceTimestampStringList.size() != 0;
float speedUp = extractor.singleValueParameter("speed", Float.class);
@@ -209,8 +205,6 @@ public class FileStreamProtocol extends Protocol {
.sourceType(AdapterSourceType.STREAM)
.category(AdapterType.Generic)
.requiredFile(Labels.withId("filePath"), Filetypes.CSV, Filetypes.JSON, Filetypes.XML)
-// .requiredSingleValueSelection(Labels.withId("replaceTimestamp"),
-// Options.from("True", "False"))
.requiredMultiValueSelection(Labels.withId("replaceTimestamp"),
Options.from(""))
.requiredFloatParameter(Labels.withId("speed"))
@@ -231,27 +225,6 @@ public class FileStreamProtocol extends Protocol {
}
}
- @Override
- public List<Map<String, Object>> getNElements(int n) throws ParseException {
- List<Map<String, Object>> result = new ArrayList<>();
-
- InputStream dataInputStream = getDataFromEndpoint();
-
- List<byte[]> dataByteArray = parser.parseNEvents(dataInputStream, n);
-
- // Check that result size is n. Currently, just an error is logged. Maybe change to an exception
- if (dataByteArray.size() < n) {
- logger.error("Error in File Protocol! User required: " + n + " elements but the resource just had: "
- + dataByteArray.size());
- }
-
- for (byte[] b : dataByteArray) {
- result.add(format.parse(b));
- }
-
- return result;
- }
-
@Override
public String getId() {
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/HttpServerProtocol.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/HttpServerProtocol.java
index e64d52885..0ba3877f0 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/HttpServerProtocol.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/HttpServerProtocol.java
@@ -49,8 +49,6 @@ import org.apache.streampipes.sdk.utils.Datatypes;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.List;
-import java.util.Map;
public class HttpServerProtocol extends Protocol {
@@ -153,11 +151,6 @@ public class HttpServerProtocol extends Protocol {
}
}
- @Override
- public List<Map<String, Object>> getNElements(int n) throws ParseException {
- return null;
- }
-
@Override
public void run(IAdapterPipeline adapterPipeline) {
SendToPipeline stk = new SendToPipeline(format, adapterPipeline);
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/HttpStreamProtocol.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/HttpStreamProtocol.java
index 735917e5a..087ed2720 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/HttpStreamProtocol.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/HttpStreamProtocol.java
@@ -39,9 +39,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.InputStream;
-import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
public class HttpStreamProtocol extends PullProtocol {
@@ -51,7 +49,6 @@ public class HttpStreamProtocol extends PullProtocol {
private static final String URL_PROPERTY = "url";
private static final String INTERVAL_PROPERTY = "interval";
- private static final String ACCESS_TOKEN_PROPERTY = "access_token";
private String url;
private String accessToken;
@@ -93,8 +90,6 @@ public class HttpStreamProtocol extends PullProtocol {
.category(AdapterType.Generic)
.requiredTextParameter(Labels.withId(URL_PROPERTY))
.requiredIntegerParameter(Labels.withId(INTERVAL_PROPERTY))
- //.requiredTextParameter(Labels.from(ACCESS_TOKEN_PROPERTY, "Access Token", "Http
- // Access Token"))
.build();
}
@@ -117,28 +112,6 @@ public class HttpStreamProtocol extends PullProtocol {
return result;
}
- @Override
- public List<Map<String, Object>> getNElements(int n) throws ParseException {
- List<Map<String, Object>> result = new ArrayList<>();
-
- InputStream dataInputStream = getDataFromEndpoint();
-
- List<byte[]> dataByte = parser.parseNEvents(dataInputStream, n);
-
- // Check that result size is n. Currently just an error is logged. Maybe change to an exception
- if (dataByte.size() < n) {
- logger.error("Error in HttpStreamProtocol! User required: " + n + " elements but the resource just had: "
- + dataByte.size());
- }
-
- for (byte[] b : dataByte) {
- result.add(format.parse(b));
- }
-
- return result;
- }
-
-
@Override
public String getId() {
return ID;
@@ -160,16 +133,9 @@ public class HttpStreamProtocol extends PullProtocol {
result = request
.execute().returnContent().asStream();
-// if (s.startsWith("ï")) {
-// s = s.substring(3);
-// }
-
-// result = IOUtils.toInputStream(s, "UTF-8");
-
} catch (Exception e) {
logger.error("Error while fetching data from URL: " + url, e);
throw new ParseException("Error while fetching data from URL: " + url);
-// throw new AdapterException();
}
if (result == null) {
throw new ParseException("Could not receive Data from file: " + url);