android - 如何使用 Android Studio 将数据从 Raspberry pi 发送到 Google Cloud IoT Core？

Question

我正在做 Android Things 项目。

我想向 Google Cloud IoT Core 发布一条字符串消息，但显示了错误。

我正在使用带有 Android Things OS 的 Raspberry Pi 3，并使用 Android Studio 对其进行编程。

错误截图：

这是整个代码：

AndroidManifest.xml：

<?xml version="1.0" encoding="utf-8"?>
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
          package="cacaosd.com.sample1">

    <!-- PAHO Permissions -->
    <uses-permission android:name="android.permission.WAKE_LOCK" />
    <uses-permission android:name="android.permission.WRITE_EXTERNAL_STORAGE" />
    <uses-permission android:name="android.permission.READ_EXTERNAL_STORAGE" />
    <uses-permission android:name="android.permission.ACCESS_NETWORK_STATE" />
    <uses-permission android:name="android.permission.INTERNET" />
    <!-- PAHO Permissions -->

    <application>
        <uses-library android:name="com.google.android.things"/>

        <!-- Mqtt Service -->
        <service android:name="org.eclipse.paho.android.service.MqttService" />

        <activity android:name=".MainActivity">
            <intent-filter>
                <action android:name="android.intent.action.MAIN"/>

                <category android:name="android.intent.category.LAUNCHER"/>
            </intent-filter>
            <intent-filter>
                <action android:name="android.intent.action.MAIN"/>

                <category android:name="android.intent.category.IOT_LAUNCHER"/>
                <category android:name="android.intent.category.DEFAULT"/>
            </intent-filter>
        </activity>
    </application>

</manifest>

IotCoreCommunicator 类

  package cacaosd.com.sample1;

import android.content.Context;
import android.util.Log;

import java.util.concurrent.TimeUnit;

import org.eclipse.paho.android.service.MqttAndroidClient;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

public class IotCoreCommunicator {

    private static final String SERVER_URI = "ssl://mqtt.googleapis.com:8883";

    public static class Builder {

        private Context context;
        private String projectId;
        private String cloudRegion;
        private String registryId;
        private String deviceId;
        private int privateKeyRawFileId;

        public Builder withContext(Context context) {
            this.context = context;
            return this;
        }

        public Builder withProjectId(String projectId) {
            this.projectId = projectId;
            return this;
        }

        public Builder withCloudRegion(String cloudRegion) {
            this.cloudRegion = cloudRegion;
            return this;
        }

        public Builder withRegistryId(String registryId) {
            this.registryId = registryId;
            return this;
        }

        public Builder withDeviceId(String deviceId) {
            this.deviceId = deviceId;
            return this;
        }

        public Builder withPrivateKeyRawFileId(int privateKeyRawFileId) {
            this.privateKeyRawFileId = privateKeyRawFileId;
            return this;
        }

        public IotCoreCommunicator build() {
            if (context == null) {
                throw new IllegalStateException("context must not be null");
            }

            if (projectId == null) {
                throw new IllegalStateException("projectId must not be null");
            }
            if (cloudRegion == null) {
                throw new IllegalStateException("cloudRegion must not be null");
            }
            if (registryId == null) {
                throw new IllegalStateException("registryId must not be null");
            }
            if (deviceId == null) {
                throw new IllegalStateException("deviceId must not be null");
            }
            String clientId = "projects/" + projectId + "/locations/" + cloudRegion + "/registries/" + registryId + "/devices/" + deviceId;

            if (privateKeyRawFileId == 0) {
                throw new IllegalStateException("privateKeyRawFileId must not be 0");
            }
            MqttAndroidClient client = new MqttAndroidClient(context, SERVER_URI, clientId);
            IotCorePasswordGenerator passwordGenerator = new IotCorePasswordGenerator(projectId, context.getResources(), privateKeyRawFileId);
            return new IotCoreCommunicator(client, deviceId, passwordGenerator);
        }

    }

    private final MqttAndroidClient client;
    private final String deviceId;
    private final IotCorePasswordGenerator passwordGenerator;

    IotCoreCommunicator(MqttAndroidClient client, String deviceId, IotCorePasswordGenerator passwordGenerator) {
        this.client = client;
        this.deviceId = deviceId;
        this.passwordGenerator = passwordGenerator;
    }

    public void connect() {
        monitorConnection();
        clientConnect();
        subscribeToConfigChanges();
    }

    private void monitorConnection() {
        client.setCallback(new MqttCallback() {
            @Override
            public void connectionLost(Throwable cause) {
                Log.e("TUT", "connection lost", cause);
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) throws Exception {
                Log.d("TUT", "message arrived " + topic + " MSG " + message);
                // You need to do something with messages when they arrive
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                Log.d("TUT", "delivery complete " + token);
            }
        });
    }

    private void clientConnect() {
        try {
            MqttConnectOptions connectOptions = new MqttConnectOptions();
            // Note that the the Google Cloud IoT Core only supports MQTT 3.1.1, and Paho requires that we explicitly set this.
            // If you don't, 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-but-necessary");
            connectOptions.setPassword(passwordGenerator.createJwtRsaPassword());

            IMqttToken iMqttToken = client.connect(connectOptions);
            iMqttToken.setActionCallback(new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    Log.d("TUT", "success, connected");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    Log.e("TUT", "failure, not connected", exception);
                }
            });
            iMqttToken.waitForCompletion(TimeUnit.SECONDS.toMillis(30));
            Log.d("TUT", "IoT Core connection established.");
        } catch (MqttException e) {
            throw new IllegalStateException(e);
        }
    }

    /**
     * Configuration is managed and sent from the IoT Core Platform
     */
    private void subscribeToConfigChanges() {
        try {
            client.subscribe("/devices/" + deviceId + "/config", 1);
        } catch (MqttException e) {
            throw new IllegalStateException(e);
        }
    }

    public void publishMessage(String subtopic, String message) {
        String topic = "/devices/" + deviceId + "/" + subtopic;
        String payload = "{msg:\"" + message + "\"}";
        MqttMessage mqttMessage = new MqttMessage(payload.getBytes());
        mqttMessage.setQos(1);
        try {
            client.publish(topic, mqttMessage);
            Log.d("TUT", "IoT Core message published. To topic: " + topic);
        } catch (MqttException e) {
            throw new IllegalStateException(e);
        }
    }

    public void disconnect() {
        try {
            Log.d("TUT", "IoT Core connection disconnected.");
            client.disconnect();
        } catch (MqttException e) {
            throw new IllegalStateException(e);
        }
    }

}

IotCorePasswordGenerator 类

    package cacaosd.com.sample1;

import android.content.res.Resources;
import android.util.Base64;

import java.io.IOException;
import java.io.InputStream;
import java.security.KeyFactory;
import java.security.NoSuchAlgorithmException;
import java.security.spec.InvalidKeySpecException;
import java.security.spec.PKCS8EncodedKeySpec;
import java.time.Duration;
import java.time.Instant;
import java.util.Date;

import io.jsonwebtoken.JwtBuilder;
import io.jsonwebtoken.Jwts;
import io.jsonwebtoken.SignatureAlgorithm;

class IotCorePasswordGenerator {

    private final String projectId;
    private final Resources resources;
    private final int privateKeyRawFileId;

    IotCorePasswordGenerator(String projectId, Resources resources, int privateKeyRawFileId) {
        this.projectId = projectId;
        this.resources = resources;
        this.privateKeyRawFileId = privateKeyRawFileId;
    }

    char[] createJwtRsaPassword() {
        try {
            byte[] privateKeyBytes = decodePrivateKey(resources, privateKeyRawFileId);
            return createJwtRsaPassword(projectId, privateKeyBytes).toCharArray();
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("Algorithm not supported. (developer error)", e);
        } catch (InvalidKeySpecException e) {
            throw new IllegalStateException("Invalid Key spec. (developer error)", e);
        } catch (IOException e) {
            throw new IllegalStateException("Cannot read private key file.", e);
        }
    }

    private static byte[] decodePrivateKey(Resources resources, int privateKeyRawFileId) throws IOException {
        try(InputStream inStream = resources.openRawResource(privateKeyRawFileId)) {
            return Base64.decode(inputToString(inStream), Base64.DEFAULT);
        }
    }

    private static String inputToString(InputStream is) {
        java.util.Scanner s = new java.util.Scanner(is).useDelimiter("\\A");
        return s.hasNext() ? s.next() : "";
    }

    private static String createJwtRsaPassword(String projectId, byte[] privateKeyBytes) throws NoSuchAlgorithmException, InvalidKeySpecException {
        return createPassword(projectId, privateKeyBytes, "RSA", SignatureAlgorithm.RS256);
    }

    private static String createPassword(String projectId, byte[] privateKeyBytes, String algorithmName, SignatureAlgorithm signatureAlgorithm) throws NoSuchAlgorithmException, InvalidKeySpecException {
        Instant now = Instant.now();
        // 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(Date.from(now))
                        .setExpiration(Date.from(now.plus(Duration.ofMinutes(20))))
                        .setAudience(projectId);

        PKCS8EncodedKeySpec spec = new PKCS8EncodedKeySpec(privateKeyBytes);
        KeyFactory kf = KeyFactory.getInstance(algorithmName);

        return jwtBuilder.signWith(signatureAlgorithm, kf.generatePrivate(spec)).compact();
    }

}

MainActivity 类：

    package cacaosd.com.sample1;

import android.app.Activity;
import android.hardware.SensorEvent;
import android.os.Bundle;
import android.os.HandlerThread;
import android.os.Handler;

import android.app.Activity;
import android.os.Bundle;
import android.os.Handler;
import android.os.HandlerThread;
import android.util.Log;



import cacaosd.com.sample1.R;
import cacaosd.com.sample1.IotCoreCommunicator;


import com.google.android.things.pio.Gpio;


import java.io.IOException;
import java.util.concurrent.TimeUnit;


public class MainActivity extends Activity {



    private IotCoreCommunicator communicator;
    private Handler handler;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        // Setup the communication with your Google IoT Core details
        communicator = new IotCoreCommunicator.Builder()
                .withContext(this)
                .withCloudRegion("us-central1") // ex: europe-west1
                .withProjectId("my-first-project-198704")   // ex: supercoolproject23236
                .withRegistryId("vibration") // ex: my-devices
                .withDeviceId("my-device") // ex: my-test-raspberry-pi
                .withPrivateKeyRawFileId(R.raw.rsa_private)
                .build();

        HandlerThread thread = new HandlerThread("MyBackgroundThread");
        thread.start();
        handler = new Handler(thread.getLooper());
        handler.post(connectOffTheMainThread); // Use whatever threading mechanism you want
    }

    private final Runnable connectOffTheMainThread = new Runnable() {
        @Override
        public void run() {
            communicator.connect();

            handler.post(sendMqttMessage);
        }
    };



    private final Runnable sendMqttMessage = new Runnable() {
        private int i;

        /**
         * We post 100 messages as an example, 1 a second
         */
        @Override
        public void run() {
            if (i == 100) {
                return;
            }



            // events is the default topic for MQTT communication
            String subtopic = "events";
            // Your message you want to send
            String message = "Hello World " + i++;

            communicator.publishMessage(subtopic, message);



            handler.postDelayed(this, TimeUnit.SECONDS.toMillis(1));
        }
    };

    @Override
    protected void onDestroy() {
        communicator.disconnect();
        super.onDestroy();
    }
}

更新：

我按照this documentation和this demo将私钥从“pem”格式转换为“pkcs8”格式，然后错误“Invalid key spec”消失了，但仍然存在“FATAL EXCEPTION”和“java.lang.IllegalArgumentException” : bad base-64" 如下图所示：

[![在此处输入图像描述][4]][4]

它说这些是导致错误的相关代码（在上图中以蓝色显示：

IotCorePasswordGenerator.java:47

return Base64.decode(inputToString(inStream), Base64.DEFAULT);

IotCorePasswordGenerator.java:34

 byte[] privateKeyBytes = decodePrivateKey(resources, privateKeyRawFileId);

IotCoreCommunicator.java:135

connectOptions.setPassword(passwordGenerator.createJwtRsaPassword());

IotCoreCommunicator.java:101

clientConnect();

MainActivity.java:58

communicator.connect();

更新 2

我删除了语句“-----BEGIN PRIVATE KEY-----”和语句“-----END PRIVATE KEY-----”，错误“bad base 64”消失了，现在还有另一个错误是“管道损坏”，如下图所示，当我重新打开 Android Studio 并重建项目时，这个错误“管道损坏”被删除，当我再次运行项目时它又回来了。

错误（第一张图片）

带有开始和结束语句的私钥（第二张图片）

没有开始和结束语句的私钥（第三张图片）

Answer 1

从错误中可以看出，您使用错误类型的 SSL 密钥注册了设备。验证您创建的 SSL 密钥是否与您在 IoT Core 中指定的格式相匹配。即，如果您使用 x509 证书包装器创建了 RSA 密钥，请确保您的设备已使用该类型注册，而不仅仅是 RSA 密钥。

还要确保私钥实际上在设备上，并且没有损坏。

编辑：问题可能是roots.pem 不在设备上处理与IoT Core 的TLS 握手。我们将看到... 要获取它，运行：wget https://pki.google.com/roots.pem并将roots.pem 与设备上的私钥放在同一目录中。

Answer 2

我强烈建议您查看适用于 Cloud IoT 核心的 Android Things 连接器。该项目使从 Android Things 访问 Cloud IoT 核心变得更加容易，并处理各种最佳实践，例如令牌刷新。