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/06 21:09:20 UTC
[incubator-plc4x] 01/03: Implement basic example of connecting an
S7 device to Google Cloud IoT Core
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 6fb35bf72fe77ea0d22faa95f1b7f93db43c18ff
Author: Veronika Schindler <ve...@tty42.de>
AuthorDate: Tue Aug 28 12:05:31 2018 +0200
Implement basic example of connecting an S7 device to Google Cloud IoT Core
---
examples/google/README.md | 61 +++++
examples/google/pom.xml | 118 +++++++++
.../google/iotcore/MqttExampleOptions.java | 166 ++++++++++++
.../google/iotcore/S7PlcToGoogleIoTCoreSample.java | 292 +++++++++++++++++++++
examples/google/src/main/resources/logback.xml | 40 +++
examples/pom.xml | 1 +
6 files changed, 678 insertions(+)
diff --git a/examples/google/README.md b/examples/google/README.md
new file mode 100644
index 0000000..42f5f71
--- /dev/null
+++ b/examples/google/README.md
@@ -0,0 +1,61 @@
+<!--
+
+ 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.
+
+-->
+# Plc to Goolgle IoT Core Adapter
+
+This Adapter opens a MQTT Bridge to Google Cloud IoT Core.
+
+To run this code, you need an account to the google cloud and a project.
+You need to create a device registry and add a device to it, as described here:
+https://cloud.google.com/iot/docs/how-tos/devices
+
+Then, you can run the PlcToGoogleIoTCoreSample to connect tho google and send some values into the cloud.
+Some sample arguments:
+
+ -project_id=myprojectname
+ -registry_id=plc4x-test
+ -cloud_region=europe-west1
+ -device_id=plc4x-test-device
+ -private_key_file=../../../rsa_private_pkcs8
+ -algorithm=RS256
+
+Some documentation can be found here:
+https://github.com/GoogleCloudPlatform/java-docs-samples/tree/master/iot/api-client/manager#cloud-iot-core-java-mqtt-example
+This code was adapted from:
+https://github.com/GoogleCloudPlatform/java-docs-samples/tree/master/iot/api-client/manager#cloud-iot-core-java-mqtt-example
+
+To retrive the messages, you can subscribe to Cloud Pub/Sub.
+For example, you could install the google cloud sdk (gcloud) and run:
+
+ gcloud auth login
+ gcloud config set myprojectname
+ gcloud iot devices configs list --project=myprojectname \\
+ --region=europe-west1 \\
+ --registry=myplc4x-test-registry \\
+ --device=myplc4x-test-device \\
+ --limit=5
+ gcloud pubsub subscriptions create --topic plc4x-test-events plc4x-test-subscription
+ gcloud pubsub subscriptions pull --auto-ack plc4x-test-subscription
+
+
+To pull more than one message, use the option --limit [number]
+
+For further reference to the Cloud Pub/Sub, see:
+https://cloud.google.com/pubsub/docs/quickstart-cli
+https://cloud.google.com/sdk/gcloud/reference/alpha/pubsub/subscriptions/pull
+
diff --git a/examples/google/pom.xml b/examples/google/pom.xml
new file mode 100644
index 0000000..173c7f2
--- /dev/null
+++ b/examples/google/pom.xml
@@ -0,0 +1,118 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>examples</artifactId>
+ <groupId>org.apache.plc4x.examples</groupId>
+ <version>0.0.1-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>google</artifactId>
+ <name>Examples: Google Cloud IoT Core</name>
+ <description>
+ Implementation of a PLC4X application gathering information from
+ a PLC and making that available in an the Google Cloud IoT Core.
+ </description>
+
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-dependency-plugin</artifactId>
+ <configuration>
+ <usedDependencies>
+ <usedDependency>org.apache.plc4x:plc4j-protocol-s7</usedDependency>
+ </usedDependencies>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-api</artifactId>
+ <version>0.0.1-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-core</artifactId>
+ <version>0.0.1-SNAPSHOT</version>
+ </dependency>
+ <dependency>
+ <groupId>org.eclipse.paho</groupId>
+ <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+ <version>1.2.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.plc4x</groupId>
+ <artifactId>plc4j-protocol-s7</artifactId>
+ <version>0.0.1-SNAPSHOT</version>
+ <scope>runtime</scope>
+ </dependency>
+ <dependency>
+ <groupId>joda-time</groupId>
+ <artifactId>joda-time</artifactId>
+ <version>2.3</version>
+ <scope>compile</scope>
+ </dependency>
+ <!-- https://mvnrepository.com/artifact/io.jsonwebtoken/jjwt -->
+ <dependency>
+ <groupId>io.jsonwebtoken</groupId>
+ <artifactId>jjwt</artifactId>
+ <version>0.9.1</version>
+ </dependency>
+ <!-- https://mvnrepository.com/artifact/commons-cli/commons-cli -->
+ <dependency>
+ <groupId>commons-cli</groupId>
+ <artifactId>commons-cli</artifactId>
+ <version>1.4</version>
+ </dependency>
+ <!-- https://mvnrepository.com/artifact/org.apache.beam/beam-runners-google-cloud-dataflow-java -->
+ <dependency>
+ <groupId>org.apache.beam</groupId>
+ <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
+ <version>2.6.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>log4j-over-slf4j</artifactId>
+ <version>1.7.25</version>
+ </dependency>
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ </dependency>
+ </dependencies>
+
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.qpid</groupId>
+ <artifactId>proton-j</artifactId>
+ <version>0.18.0</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+</project>
\ No newline at end of file
diff --git a/examples/google/src/main/java/org/apache/plc4x/java/examples/google/iotcore/MqttExampleOptions.java b/examples/google/src/main/java/org/apache/plc4x/java/examples/google/iotcore/MqttExampleOptions.java
new file mode 100644
index 0000000..7cdbc9c
--- /dev/null
+++ b/examples/google/src/main/java/org/apache/plc4x/java/examples/google/iotcore/MqttExampleOptions.java
@@ -0,0 +1,166 @@
+/*
+ * Copyright 2017 Google Inc.
+ *
+ * Licensed 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.plc4x.java.examples.google.iotcore;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+
+/** Command line options for the MQTT example. */
+public class MqttExampleOptions {
+ String projectId;
+ String registryId;
+ String deviceId;
+ String privateKeyFile;
+ String algorithm;
+ String cloudRegion = "europe-west1";
+ int numMessages = 100;
+ int tokenExpMins = 20;
+ String mqttBridgeHostname = "mqtt.googleapis.com";
+ short mqttBridgePort = 8883;
+ String messageType = "event";
+
+ /** Construct an MqttExampleOptions class from command line flags. */
+ public static MqttExampleOptions fromFlags(String[] args) {
+ Options options = new Options();
+ // Required arguments
+ options.addOption(
+ Option.builder()
+ .type(String.class)
+ .longOpt("project_id")
+ .hasArg()
+ .desc("GCP cloud project name.")
+ .required()
+ .build());
+ options.addOption(
+ Option.builder()
+ .type(String.class)
+ .longOpt("registry_id")
+ .hasArg()
+ .desc("Cloud IoT Core registry id.")
+ .required()
+ .build());
+ options.addOption(
+ Option.builder()
+ .type(String.class)
+ .longOpt("device_id")
+ .hasArg()
+ .desc("Cloud IoT Core device id.")
+ .required()
+ .build());
+ options.addOption(
+ Option.builder()
+ .type(String.class)
+ .longOpt("private_key_file")
+ .hasArg()
+ .desc("Path to private key file.")
+ .required()
+ .build());
+ options.addOption(
+ Option.builder()
+ .type(String.class)
+ .longOpt("algorithm")
+ .hasArg()
+ .desc("Encryption algorithm to use to generate the JWT. Either 'RS256' or 'ES256'.")
+ .required()
+ .build());
+
+ // Optional arguments.
+ options.addOption(
+ Option.builder()
+ .type(String.class)
+ .longOpt("cloud_region")
+ .hasArg()
+ .desc("GCP cloud region.")
+ .build());
+ options.addOption(
+ Option.builder()
+ .type(Number.class)
+ .longOpt("num_messages")
+ .hasArg()
+ .desc("Number of messages to publish.")
+ .build());
+ options.addOption(
+ Option.builder()
+ .type(String.class)
+ .longOpt("mqtt_bridge_hostname")
+ .hasArg()
+ .desc("MQTT bridge hostname.")
+ .build());
+ options.addOption(
+ Option.builder()
+ .type(Number.class)
+ .longOpt("token_exp_minutes")
+ .hasArg()
+ .desc("Minutes to JWT token refresh (token expiration time).")
+ .build());
+ options.addOption(
+ Option.builder()
+ .type(Number.class)
+ .longOpt("mqtt_bridge_port")
+ .hasArg()
+ .desc("MQTT bridge port.")
+ .build());
+ options.addOption(
+ Option.builder()
+ .type(String.class)
+ .longOpt("message_type")
+ .hasArg()
+ .desc("Indicates whether the message is a telemetry event or a device state message")
+ .build());
+
+ CommandLineParser parser = new DefaultParser();
+ CommandLine commandLine;
+ try {
+ commandLine = parser.parse(options, args);
+ MqttExampleOptions res = new MqttExampleOptions();
+
+ res.projectId = commandLine.getOptionValue("project_id");
+ res.registryId = commandLine.getOptionValue("registry_id");
+ res.deviceId = commandLine.getOptionValue("device_id");
+ res.privateKeyFile = commandLine.getOptionValue("private_key_file");
+ res.algorithm = commandLine.getOptionValue("algorithm");
+ if (commandLine.hasOption("cloud_region")) {
+ res.cloudRegion = commandLine.getOptionValue("cloud_region");
+ }
+ if (commandLine.hasOption("num_messages")) {
+ res.numMessages = ((Number) commandLine.getParsedOptionValue("num_messages")).intValue();
+ }
+ if (commandLine.hasOption("token_exp_minutes")) {
+ res.tokenExpMins =
+ ((Number) commandLine.getParsedOptionValue("token_exp_minutes")).intValue();
+ }
+ if (commandLine.hasOption("mqtt_bridge_hostname")) {
+ res.mqttBridgeHostname = commandLine.getOptionValue("mqtt_bridge_hostname");
+ }
+ if (commandLine.hasOption("mqtt_bridge_port")) {
+ res.mqttBridgePort =
+ ((Number) commandLine.getParsedOptionValue("mqtt_bridge_port")).shortValue();
+ }
+ if (commandLine.hasOption("message_type")) {
+ res.messageType = commandLine.getOptionValue("message_type");
+ }
+ return res;
+ } catch (ParseException e) {
+ System.err.println(e.getMessage());
+ return null;
+ }
+ }
+}
\ No newline at end of file
diff --git a/examples/google/src/main/java/org/apache/plc4x/java/examples/google/iotcore/S7PlcToGoogleIoTCoreSample.java b/examples/google/src/main/java/org/apache/plc4x/java/examples/google/iotcore/S7PlcToGoogleIoTCoreSample.java
new file mode 100644
index 0000000..bb5a43a
--- /dev/null
+++ b/examples/google/src/main/java/org/apache/plc4x/java/examples/google/iotcore/S7PlcToGoogleIoTCoreSample.java
@@ -0,0 +1,292 @@
+/*
+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.plc4x.java.examples.google.iotcore;
+
+import org.apache.plc4x.java.PlcDriverManager;
+import org.apache.plc4x.java.api.connection.PlcConnection;
+import org.apache.plc4x.java.api.connection.PlcReader;
+import org.apache.plc4x.java.api.messages.items.ReadResponseItem;
+import org.apache.plc4x.java.api.messages.specific.TypeSafePlcReadRequest;
+import org.apache.plc4x.java.api.messages.specific.TypeSafePlcReadResponse;
+import org.apache.plc4x.java.api.model.Address;
+
+// [START iot_mqtt_includes]
+import io.jsonwebtoken.JwtBuilder;
+import io.jsonwebtoken.Jwts;
+import io.jsonwebtoken.SignatureAlgorithm;
+
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.security.KeyFactory;
+import java.security.spec.PKCS8EncodedKeySpec;
+
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallback;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+// [END iot_mqtt_includes]
+
+public class S7PlcToGoogleIoTCoreSample {
+
+ private static final Logger logger = LoggerFactory.getLogger(S7PlcToGoogleIoTCoreSample.class);
+
+ // [START iot_mqtt_jwt]
+
+ /**
+ * Create a Cloud IoT Core JWT for the given project id, signed with the given RSA key.
+ */
+ private static String createJwtRsa(String projectId, String privateKeyFile) throws Exception {
+ DateTime now = new DateTime();
+ // Create a JWT to authenticate this device. The device will be disconnected after the token
+ // expires, and will have to reconnect with a new token. The audience field should always be set
+ // to the GCP project id.
+ JwtBuilder jwtBuilder =
+ Jwts.builder()
+ .setIssuedAt(now.toDate())
+ .setExpiration(now.plusMinutes(20).toDate())
+ .setAudience(projectId);
+
+ byte[] keyBytes = Files.readAllBytes(Paths.get(privateKeyFile));
+ PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(keyBytes);
+ KeyFactory kf = KeyFactory.getInstance("RSA");
+
+ return jwtBuilder.signWith(SignatureAlgorithm.RS256, kf.generatePrivate(spec)).compact();
+ }
+
+ /**
+ * Create a Cloud IoT Core JWT for the given project id, signed with the given ES key.
+ */
+ private static String createJwtEs(String projectId, String privateKeyFile) throws Exception {
+ DateTime now = new DateTime();
+ // Create a JWT to authenticate this device. The device will be disconnected after the token
+ // expires, and will have to reconnect with a new token. The audience field should always be set
+ // to the GCP project id.
+ JwtBuilder jwtBuilder =
+ Jwts.builder()
+ .setIssuedAt(now.toDate())
+ .setExpiration(now.plusMinutes(20).toDate())
+ .setAudience(projectId);
+
+ byte[] keyBytes = Files.readAllBytes(Paths.get(privateKeyFile));
+ PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(keyBytes);
+ KeyFactory kf = KeyFactory.getInstance("EC");
+
+ return jwtBuilder.signWith(SignatureAlgorithm.ES256, kf.generatePrivate(spec)).compact();
+ }
+ // [END iot_mqtt_jwt]
+
+ /**
+ * Attaches the callback used when configuration changes occur.
+ */
+ private static void attachCallback(MqttClient client, String deviceId) throws MqttException {
+ // [START iot_mqtt_configcallback]
+ MqttCallback mCallback = new MqttCallback() {
+ @Override
+ public void connectionLost(Throwable cause) {
+ // Do nothing...
+ }
+
+ @Override
+ public void messageArrived(String topic, MqttMessage message) {
+ String payload = new String(message.getPayload());
+ System.out.println("Payload : " + payload);
+ // TODO: Insert your parsing / handling of the configuration message here.
+ }
+
+ @Override
+ public void deliveryComplete(IMqttDeliveryToken token) {
+ // Do nothing;
+ }
+ };
+
+ String configTopic = String.format("/devices/%s/config", deviceId);
+ client.subscribe(configTopic, 1);
+
+ client.setCallback(mCallback);
+ }
+ // [END iot_mqtt_configcallback]
+
+
+ private static void setConnectPassword(MqttExampleOptions options, MqttConnectOptions connectOptions) throws Exception {
+ switch (options.algorithm) {
+ case "RS256":
+ connectOptions.setPassword(
+ createJwtRsa(options.projectId, options.privateKeyFile).toCharArray());
+ break;
+ case "ES256":
+ connectOptions.setPassword(
+ createJwtEs(options.projectId, options.privateKeyFile).toCharArray());
+ break;
+ default:
+ throw new IllegalArgumentException(
+ "Invalid algorithm " + options.algorithm
+ + ". Should be one of 'RS256' or 'ES256'.");
+ }
+ }
+
+ /**
+ * Example code do demonstrate sending events from an S7 device to Microsoft Azure IoT Hub
+ *
+ * @param args Expected: [plc4x connection string, plc4x address, IoT-Hub connection string].
+ */
+ public static void main(String[] args) throws Exception {
+
+ // [START iot_mqtt_configuremqtt]
+ MqttExampleOptions options = MqttExampleOptions.fromFlags(args);
+ if (options == null) {
+ // Could not parse.
+ System.exit(1);
+ }
+
+ // Build the connection string for Google's Cloud IoT Core MQTT server. Only SSL
+ // connections are accepted. For server authentication, the JVM's root certificates
+ // are used.
+ final String mqttServerAddress =
+ String.format("ssl://%s:%s", options.mqttBridgeHostname, options.mqttBridgePort);
+
+ // Create our MQTT client. The mqttClientId is a unique string that identifies this device. For
+ // Google Cloud IoT Core, it must be in the format below.
+ final String mqttClientId =
+ String.format("projects/%s/locations/%s/registries/%s/devices/%s",
+ options.projectId, options.cloudRegion, options.registryId, options.deviceId);
+
+ MqttConnectOptions connectOptions = new MqttConnectOptions();
+ // Note that the Google Cloud IoT Core only supports MQTT 3.1.1, and Paho requires that we
+ // explictly set this. If you don't set MQTT version, the server will immediately close its
+ // connection to your device.
+ connectOptions.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
+
+ // With Google Cloud IoT Core, the username field is ignored, however it must be set for the
+ // Paho client library to send the password field. The password field is used to transmit a JWT
+ // to authorize the device.
+ connectOptions.setUserName("unused");
+
+ DateTime iat = new DateTime();
+ setConnectPassword(options, connectOptions);
+ // [END iot_mqtt_configuremqtt]
+
+ // [START iot_mqtt_publish]
+ // Create a client, and connect to the Google MQTT bridge.
+ MqttClient client = new MqttClient(mqttServerAddress, mqttClientId, new MemoryPersistence());
+
+ // Both connect and publish operations may fail. If they do, allow retries but with an
+ // exponential backoff time period.
+ long initialConnectIntervalMillis = 500L;
+ long maxConnectIntervalMillis = 6000L;
+ long maxConnectRetryTimeElapsedMillis = 900000L;
+ float intervalMultiplier = 1.5f;
+
+ long retryIntervalMs = initialConnectIntervalMillis;
+ long totalRetryTimeMs = 0;
+
+ while (!client.isConnected() && totalRetryTimeMs < maxConnectRetryTimeElapsedMillis) {
+ try {
+ client.connect(connectOptions);
+ } catch (MqttException e) {
+ int reason = e.getReasonCode();
+
+ // If the connection is lost or if the server cannot be connected, allow retries, but with
+ // exponential backoff.
+ System.out.println("An error occurred: " + e.getMessage());
+ if (reason == MqttException.REASON_CODE_CONNECTION_LOST
+ || reason == MqttException.REASON_CODE_SERVER_CONNECT_ERROR) {
+ System.out.println("Retrying in " + retryIntervalMs / 1000.0 + " seconds.");
+ Thread.sleep(retryIntervalMs);
+ totalRetryTimeMs += retryIntervalMs;
+ retryIntervalMs *= intervalMultiplier;
+ if (retryIntervalMs > maxConnectIntervalMillis) {
+ retryIntervalMs = maxConnectIntervalMillis;
+ }
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ attachCallback(client, options.deviceId);
+
+ // Publish to the events or state topic based on the flag.
+ String subTopic = options.messageType.equals("event") ? "events" : options.messageType;
+
+ // The MQTT topic that this device will publish telemetry data to. The MQTT topic name is
+ // required to be in the format below. Note that this is not the same as the device registry's
+ // Cloud Pub/Sub topic.
+ String mqttTopic = String.format("/devices/%s/%s", options.deviceId, subTopic);
+
+ // Connect to Plc
+ logger.info("Connecting to Plc");
+ try (PlcConnection plcConnection = new PlcDriverManager().getConnection("s7://10.10.64.20/1/1")) {
+ logger.info("Connected");
+
+ PlcReader plcReader = plcConnection.getReader().orElseThrow(IllegalAccessError::new);
+
+ Address outputs = plcConnection.parseAddress("OUTPUTS/0");
+
+ while (!Thread.currentThread().isInterrupted()) {
+
+ TypeSafePlcReadResponse<Byte> plcReadResponse = plcReader.read(
+ new TypeSafePlcReadRequest<>(Byte.class, outputs)).get();
+ logger.info(plcReadResponse.getResponseItem().get().getValues().get(0).toString());
+
+ // Refresh the connection credentials before the JWT expires.
+ // [START iot_mqtt_jwt_refresh]
+ long secsSinceRefresh = ((new DateTime()).getMillis() - iat.getMillis()) / 1000;
+ if (secsSinceRefresh > (options.tokenExpMins * 60)) {
+ System.out.format("\tRefreshing token after: %d seconds\n", secsSinceRefresh);
+ iat = new DateTime();
+ setConnectPassword(options, connectOptions);
+ client.disconnect();
+ client.connect();
+ attachCallback(client, options.deviceId);
+ }
+ // [END iot_mqtt_jwt_refresh]
+
+ // Send data to cloud
+ for (ReadResponseItem<Byte> responseItem : plcReadResponse.getResponseItems()) {
+ Long l = responseItem.getValues().get(0).longValue();
+ byte[] array = ByteBuffer.allocate(8).putLong(l).array();
+ String result = Long.toBinaryString(l);
+ System.out.println("Outputs: " + result);
+ // Publish "array" to the MQTT topic. qos=1 means at least once delivery. Cloud IoT Core
+ // also supports qos=0 for at most once delivery.
+ MqttMessage message = new MqttMessage(array);
+ message.setQos(1);
+ client.publish(mqttTopic, message);
+ if (options.messageType.equals("event")) {
+ // Send telemetry events every second
+ Thread.sleep(1000);
+ } else {
+ // Note: Update Device state less frequently than with telemetry events
+ Thread.sleep(5000);
+ }
+ }
+ }
+ }
+
+ System.out.println("Sent all messages. Goodbye!");
+ // [END iot_mqtt_publish]
+ }
+}
diff --git a/examples/google/src/main/resources/logback.xml b/examples/google/src/main/resources/logback.xml
new file mode 100644
index 0000000..0fd17ea
--- /dev/null
+++ b/examples/google/src/main/resources/logback.xml
@@ -0,0 +1,40 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ 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.
+
+-->
+<configuration xmlns="http://ch.qos.logback/xml/ns/logback"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://ch.qos.logback/xml/ns/logback https://raw.githubusercontent.com/enricopulatzo/logback-XSD/master/src/main/xsd/logback.xsd">
+
+ <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+ <!-- encoders are assigned the type
+ ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+ <encoder>
+ <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+ </encoder>
+ </appender>
+
+ <!--<logger level="OFF" name="io.netty.util" additivity="false">-->
+ <!--<appender-ref ref="STDOUT" />-->
+ <!--</logger>-->
+
+ <root level="info">
+ <appender-ref ref="STDOUT" />
+ </root>
+
+</configuration>
\ No newline at end of file
diff --git a/examples/pom.xml b/examples/pom.xml
index 5e63794..426dc6f 100644
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@ -42,6 +42,7 @@
<module>kafka-bridge</module>
<module>plclogger</module>
<module>azure</module>
+ <module>google</module>
</modules>
<dependencyManagement>