You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by cd...@apache.org on 2018/09/17 07:07:55 UTC
[incubator-plc4x] 11/29: progress on subscription support
This is an automated email from the ASF dual-hosted git repository.
cdutz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-plc4x.git
commit 910d379768c034b405becff8d7e23f2a1523820b
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Thu Sep 13 11:38:32 2018 +0200
progress on subscription support
---
.../java/org/apache/plc4x/camel/Plc4XConsumer.java | 13 ++---
.../java/org/apache/plc4x/camel/MockDriver.java | 16 ++----
.../plc4x/java/api/connection/PlcSubscriber.java | 37 +++++++++++-
.../api/messages/PlcUnsubscriptionRequest.java | 15 +++--
.../messages/DefaultPlcUnsubscriptionRequest.java | 65 ++++++++++------------
5 files changed, 83 insertions(+), 63 deletions(-)
diff --git a/integrations/apache-camel/src/main/java/org/apache/plc4x/camel/Plc4XConsumer.java b/integrations/apache-camel/src/main/java/org/apache/plc4x/camel/Plc4XConsumer.java
index f2cccee..42544f5 100644
--- a/integrations/apache-camel/src/main/java/org/apache/plc4x/camel/Plc4XConsumer.java
+++ b/integrations/apache-camel/src/main/java/org/apache/plc4x/camel/Plc4XConsumer.java
@@ -26,7 +26,10 @@ import org.apache.camel.util.AsyncProcessorConverterHelper;
import org.apache.plc4x.java.api.connection.PlcConnection;
import org.apache.plc4x.java.api.connection.PlcSubscriber;
import org.apache.plc4x.java.api.exceptions.PlcException;
-import org.apache.plc4x.java.api.messages.*;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionEvent;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse;
+import org.apache.plc4x.java.api.messages.PlcUnsubscriptionResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -92,13 +95,7 @@ public class Plc4XConsumer extends ServiceSupport implements Consumer, java.util
protected void doStop() throws InterruptedException, ExecutionException, TimeoutException, PlcException {
PlcSubscriber plcSubscriber = plcConnection.getSubscriber().orElseThrow(
() -> new PlcException("Connection doesn't support subscriptions."));
- PlcUnsubscriptionRequest.Builder builder = plcSubscriber.unsubscriptionRequestBuilder();
- // For every field we subscribed for, now unsubscribe.
- subscriptionResponse.getFieldNames().forEach(fieldName -> {
- builder.addField(fieldName, subscriptionResponse.getSubscriptionHandle(fieldName));
- });
- PlcUnsubscriptionRequest request = builder.build();
- CompletableFuture<PlcUnsubscriptionResponse> unsubscriptionFuture = plcSubscriber.unsubscribe(request);
+ CompletableFuture<PlcUnsubscriptionResponse> unsubscriptionFuture = plcSubscriber.unsubscribe(builder -> builder.addHandles(subscriptionResponse.getSubscriptionHandles()));
PlcUnsubscriptionResponse unsubscriptionResponse = unsubscriptionFuture.get(5, TimeUnit.SECONDS);
// TODO: Handle the response ...
try {
diff --git a/integrations/apache-camel/src/test/java/org/apache/plc4x/camel/MockDriver.java b/integrations/apache-camel/src/test/java/org/apache/plc4x/camel/MockDriver.java
index b8575eb..fb99abe 100644
--- a/integrations/apache-camel/src/test/java/org/apache/plc4x/camel/MockDriver.java
+++ b/integrations/apache-camel/src/test/java/org/apache/plc4x/camel/MockDriver.java
@@ -23,26 +23,18 @@ import org.apache.plc4x.java.api.authentication.PlcAuthentication;
import org.apache.plc4x.java.api.connection.PlcConnection;
import org.apache.plc4x.java.api.connection.PlcSubscriber;
import org.apache.plc4x.java.api.connection.PlcWriter;
-import org.apache.plc4x.java.api.exceptions.PlcException;
import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse;
-import org.apache.plc4x.java.api.model.PlcField;
-import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
-import org.apache.plc4x.java.api.types.PlcResponseCode;
import org.apache.plc4x.java.base.messages.DefaultPlcSubscriptionResponse;
+import org.apache.plc4x.java.base.messages.InternalPlcSubscriptionRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Calendar;
-import java.util.Collections;
-import java.util.List;
+import java.util.HashMap;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
import static org.mockito.Mockito.*;
@@ -70,7 +62,7 @@ public class MockDriver implements PlcDriver {
// Mock a typical subscriber.
PlcSubscriber plcSubscriber = mock(PlcSubscriber.class, RETURNS_DEEP_STUBS);
- when(plcSubscriber.subscribe(any())).thenAnswer(invocation -> {
+ when(plcSubscriber.subscribe(any(PlcSubscriptionRequest.class))).thenAnswer(invocation -> {
LOGGER.info("Received {}", invocation);
// TODO: Translate this so it actually does something ...
/*PlcSubscriptionRequest subscriptionRequest = invocation.getArgument(0);
@@ -93,7 +85,7 @@ public class MockDriver implements PlcDriver {
mock(PlcSubscriptionHandle.class, RETURNS_DEEP_STUBS), PlcResponseCode.OK);
}).collect(Collectors.toList());
PlcSubscriptionResponse response = new PlcSubscriptionResponse(subscriptionRequest, responseItems);*/
- PlcSubscriptionResponse response = new DefaultPlcSubscriptionResponse();
+ PlcSubscriptionResponse response = new DefaultPlcSubscriptionResponse(mock(InternalPlcSubscriptionRequest.class), new HashMap<>());
CompletableFuture<PlcSubscriptionResponse> responseFuture = new CompletableFuture<>();
responseFuture.complete(response);
return responseFuture;
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcSubscriber.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcSubscriber.java
index 9cca159..0581bd6 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcSubscriber.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/connection/PlcSubscriber.java
@@ -22,6 +22,8 @@ import org.apache.plc4x.java.api.messages.*;
import org.apache.plc4x.java.api.model.PlcConsumerRegistration;
import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Consumer;
@@ -78,16 +80,45 @@ public interface PlcSubscriber {
*
* @param subscriptionRequest subscription request
* @param consumer consumer for all {@link PlcSubscriptionEvent}s
+ * @return TODO: document me
* @throws ExecutionException something went wrong.
* @throws InterruptedException something went wrong.
*/
- default void register(PlcSubscriptionRequest subscriptionRequest, Consumer<PlcSubscriptionEvent> consumer) throws ExecutionException, InterruptedException {
+ default PlcConsumerRegistration register(PlcSubscriptionRequest subscriptionRequest, Consumer<PlcSubscriptionEvent> consumer) throws ExecutionException, InterruptedException {
PlcSubscriptionResponse plcSubscriptionResponse = subscribe(subscriptionRequest).get();
- register(consumer, plcSubscriptionResponse.getSubscriptionHandles().toArray(new PlcSubscriptionHandle[0]));
+ // TODO: we need to return the plcSubscriptionResponse here too as we need this to unsubscribe...
+ return register(consumer, plcSubscriptionResponse.getSubscriptionHandles().toArray(new PlcSubscriptionHandle[0]));
}
- PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent> consumer, PlcSubscriptionHandle... handles);
+ /**
+ * Convenience method to subscribe a {@link Consumer} to all fields of the subscription.
+ *
+ * @param subscriptionRequestBuilderConsumer consumer for building subscription request.
+ * @param consumer consumer for all {@link PlcSubscriptionEvent}s
+ * @return TODO: document me
+ * @throws ExecutionException something went wrong.
+ * @throws InterruptedException something went wrong.
+ */
+ default PlcConsumerRegistration register(Consumer<PlcSubscriptionRequest.Builder> subscriptionRequestBuilderConsumer, Consumer<PlcSubscriptionEvent> consumer) throws ExecutionException, InterruptedException {
+ PlcSubscriptionRequest.Builder builder = subscriptionRequestBuilder();
+ subscriptionRequestBuilderConsumer.accept(builder);
+ return register(builder.build(), consumer);
+ }
+ /**
+ * @param consumer
+ * @param handles
+ * @return TODO: document me
+ */
+ PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent> consumer, Collection<PlcSubscriptionHandle> handles);
+
+ default PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent> consumer, PlcSubscriptionHandle... handles) {
+ return register(consumer, Arrays.asList(handles));
+ }
+
+ /**
+ * // TODO: document me.
+ */
void unregister(PlcConsumerRegistration registration);
PlcSubscriptionRequest.Builder subscriptionRequestBuilder();
diff --git a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcUnsubscriptionRequest.java b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcUnsubscriptionRequest.java
index ba5e065..ac7b66a 100644
--- a/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcUnsubscriptionRequest.java
+++ b/plc4j/api/src/main/java/org/apache/plc4x/java/api/messages/PlcUnsubscriptionRequest.java
@@ -20,17 +20,22 @@ package org.apache.plc4x.java.api.messages;
import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
+import java.util.Collection;
+
public interface PlcUnsubscriptionRequest extends PlcFieldRequest {
interface Builder extends PlcMessageBuilder<PlcUnsubscriptionRequest> {
/**
- * Adds a new field to the to be constructed request which should be polled cyclically.
+ * TODO document me:
*
- * @param name alias of the field.
- * @param handle subscription handle containing information about the subscription.
- * @return builder.
+ * @param plcSubscriptionHandle
+ * @return
*/
- PlcUnsubscriptionRequest.Builder addField(String name, PlcSubscriptionHandle handle);
+ PlcUnsubscriptionRequest.Builder addHandle(PlcSubscriptionHandle plcSubscriptionHandle);
+
+ PlcUnsubscriptionRequest.Builder addHandle(PlcSubscriptionHandle plcSubscriptionHandle1, PlcSubscriptionHandle... plcSubscriptionHandles);
+
+ PlcUnsubscriptionRequest.Builder addHandles(Collection<PlcSubscriptionHandle> plcSubscriptionHandle);
}
}
diff --git a/plc4j/protocols/driver-bases/base/src/main/java/org/apache/plc4x/java/base/messages/DefaultPlcUnsubscriptionRequest.java b/plc4j/protocols/driver-bases/base/src/main/java/org/apache/plc4x/java/base/messages/DefaultPlcUnsubscriptionRequest.java
index cfbf1e1..3f7fee5 100644
--- a/plc4j/protocols/driver-bases/base/src/main/java/org/apache/plc4x/java/base/messages/DefaultPlcUnsubscriptionRequest.java
+++ b/plc4j/protocols/driver-bases/base/src/main/java/org/apache/plc4x/java/base/messages/DefaultPlcUnsubscriptionRequest.java
@@ -18,84 +18,79 @@
*/
package org.apache.plc4x.java.base.messages;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
import org.apache.plc4x.java.api.messages.PlcUnsubscriptionRequest;
import org.apache.plc4x.java.api.model.PlcField;
import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
-import org.apache.plc4x.java.base.connection.PlcFieldHandler;
-import org.apache.plc4x.java.base.messages.items.FieldItem;
import org.apache.plc4x.java.base.model.InternalPlcSubscriptionHandle;
import java.util.*;
-import java.util.function.BiFunction;
+import java.util.stream.Collectors;
public class DefaultPlcUnsubscriptionRequest implements InternalPlcUnsubscriptionRequest {
+ private final Collection<? extends InternalPlcSubscriptionHandle> internalPlcSubscriptionHandles;
+
+ public DefaultPlcUnsubscriptionRequest(Collection<? extends InternalPlcSubscriptionHandle> internalPlcSubscriptionHandles) {
+ this.internalPlcSubscriptionHandles = internalPlcSubscriptionHandles;
+ }
+
@Override
public int getNumberOfFields() {
- return 0;
+ throw new IllegalStateException("not available");
}
@Override
public LinkedHashSet<String> getFieldNames() {
- return null;
+ throw new IllegalStateException("not available");
}
@Override
public PlcField getField(String name) {
- return null;
+ throw new IllegalStateException("not available");
}
@Override
public LinkedList<PlcField> getFields() {
- return null;
+ throw new IllegalStateException("not available");
}
@Override
public Collection<? extends InternalPlcSubscriptionHandle> getInternalPlcSubscriptionHandles() {
- return null;
+ return internalPlcSubscriptionHandles;
}
public static class Builder implements PlcUnsubscriptionRequest.Builder {
- private final PlcFieldHandler fieldHandler;
- private final Map<String, BuilderItem<Object>> fields;
+ private List<InternalPlcSubscriptionHandle> plcSubscriptionHandles;
- public Builder(PlcFieldHandler fieldHandler) {
- this.fieldHandler = fieldHandler;
- fields = new TreeMap<>();
+ public Builder() {
+ plcSubscriptionHandles = new ArrayList<>();
+ }
+
+ public PlcUnsubscriptionRequest.Builder addHandle(PlcSubscriptionHandle plcSubscriptionHandle) {
+ plcSubscriptionHandles.add((InternalPlcSubscriptionHandle) plcSubscriptionHandle);
+ return this;
}
@Override
- public PlcUnsubscriptionRequest.Builder addField(String name, PlcSubscriptionHandle handle) {
+ public PlcUnsubscriptionRequest.Builder addHandle(PlcSubscriptionHandle plcSubscriptionHandle1, PlcSubscriptionHandle... plcSubscriptionHandles) {
+ this.plcSubscriptionHandles.add((InternalPlcSubscriptionHandle) plcSubscriptionHandle1);
+ this.plcSubscriptionHandles.addAll(Arrays.stream(plcSubscriptionHandles).map(InternalPlcSubscriptionHandle.class::cast).collect(Collectors.toList()));
return null;
}
@Override
- public PlcUnsubscriptionRequest build() {
- LinkedHashMap<String, Pair<PlcField, FieldItem>> parsedFields = new LinkedHashMap<>();
- fields.forEach((name, builderItem) -> {
- // Compile the query string.
- PlcField parsedField = fieldHandler.createField(builderItem.fieldQuery);
- // Encode the payload.
- // TODO: Depending on the field type, handle the FieldItem creation differently.
- FieldItem fieldItem = builderItem.encoder.apply(parsedField, null);
- parsedFields.put(name, new ImmutablePair<>(parsedField, fieldItem));
- });
- return new DefaultPlcUnsubscriptionRequest();
+ public PlcUnsubscriptionRequest.Builder addHandles(Collection<PlcSubscriptionHandle> plcSubscriptionHandles) {
+ this.plcSubscriptionHandles.addAll(plcSubscriptionHandles.stream().map(InternalPlcSubscriptionHandle.class::cast).collect(Collectors.toList()));
+ return null;
}
- private static class BuilderItem<T> {
- private final String fieldQuery;
- private final BiFunction<PlcField, T[], FieldItem> encoder;
-
- private BuilderItem(String fieldQuery, BiFunction<PlcField, T[], FieldItem> encoder) {
- this.fieldQuery = fieldQuery;
- this.encoder = encoder;
- }
+ @Override
+ public PlcUnsubscriptionRequest build() {
+ return new DefaultPlcUnsubscriptionRequest(plcSubscriptionHandles);
}
+
}
}