Flutter and AWS IoT Core with Cognito: Signed WebSockets, Reconnects and Device Shadow

If your devices talk to AWS IoT Core and your app signs users in with Amazon Cognito, you'll notice a gap early. Amplify for JavaScript has a PubSub category that connects to AWS IoT in a few lines. Amplify Flutter covers auth, APIs, storage and more, but has no PubSub category. In Flutter, you build that layer yourself on top of mqtt_client.

I've built it in six production apps, and the same problems came up every time. This post walks through them, then shows aws_iot_core, the package I published so that this code is written and tested once.

How a Cognito user connects to AWS IoT Core

Devices usually connect with X.509 certificates on port 8883. A consumer app shouldn't ship a certificate, because anyone can pull it out of the APK or IPA. Instead:

The user signs in, and a Cognito identity pool hands the app temporary AWS credentials: an access key, a secret key and a session token, valid for about an hour.
The app signs a WebSocket URL with those credentials (SigV4).
It opens wss://<your-endpoint>/mqtt on port 443 and speaks MQTT 3.1.1 over it.

There are two permission layers, and missing the second one is the most common reason a connection closes right after it opens:

The identity pool's IAM role needs iot:Connect, iot:Publish, iot:Subscribe and iot:Receive.
Authenticated Cognito identities also need an AWS IoT policy attached to their identity ID (iot:AttachPolicy). This is usually done once per user by a backend endpoint or a Lambda.

Signing the URL

The URL is a SigV4 query-signed request: service iotdevicegateway, signed header host, and the hash of an empty payload. One detail differs from most AWS services: the session token is not part of the signed request. AWS IoT expects X-Amz-Security-Token to be appended after X-Amz-Signature. If you build the URL with a generic SigV4 library, check where it puts the token.

The reconnect failure that shows up after an hour

This is the bug that passes every test on your desk. mqtt_client has an auto-reconnect option, and it reconnects to the same URL it used the first time. That URL was signed with temporary credentials that expire. The app works all morning, the phone sleeps through lunch, and from then on every reconnect fails.

The fix is to own the reconnect loop instead of delegating it:

On an unexpected disconnect, wait with exponential backoff and a little jitter.
Fetch credentials again. Amplify's fetchAuthSession refreshes them when they've expired.
Sign a new URL, connect a new client, and re-subscribe every active topic.
Ignore late callbacks from the old client, so a stale "disconnected" event can't tear down the new connection.

Client IDs kick each other off

AWS IoT allows one connection per client ID. When a second connection uses the same ID, AWS drops the first. If you use the user ID as the client ID, the same account signed in on two phones will knock each other offline in a loop. Use a random ID per app instance, or make it unique per install if your IoT policy restricts client IDs.

Overlapping topics deliver duplicates

While testing the package against a real broker, I subscribed one client to both devices/lamp/# and devices/lamp/state, then published a single message to devices/lamp/state. Each listener received it twice. AWS IoT sends one copy of a message for every subscription of that client that matches it, which the MQTT spec allows.

This also catches Device Shadow code. Subscribing to .../shadow/update/+ to catch responses also matches update/delta, so a delta listener can apply the same change twice. Subscribe to the exact /accepted and /rejected topics instead.

Device Shadow is request and response over pub/sub

A shadow get is not an HTTP call. You publish to $aws/things/<thing>/shadow/get and wait for an answer on get/accepted or get/rejected. To do that reliably:

Subscribe to the response topics before publishing, or a fast response arrives before anyone is listening.
Send a clientToken with each request and match responses by it, because other clients' responses arrive on the same topics.
Add a timeout. If the subscription silently failed, nothing will ever answer.

Named shadows use the same layout under $aws/things/<thing>/shadow/name/<name>/.

The package: aws_iot_core

aws_iot_core handles all of the above. It has no Amplify dependency: you pass it a credentials provider, and it calls the provider before every connect and reconnect.

import 'package:aws_iot_core/aws_iot_core.dart';

final client = AwsIotClient(
  endpoint: 'xxxxxxxx-ats.iot.ap-south-1.amazonaws.com',
  region: 'ap-south-1',
  credentialsProvider: AmplifyIotCredentialsProvider(),
);

client.subscribe('devices/+/telemetry').listen((message) {
  print('${message.topic}: ${message.json}');
});

await client.connect();
client.publishJson('devices/lamp-1/commands', {'power': 'on'});

The Cognito credentials come from a small adapter in your app:

class AmplifyIotCredentialsProvider implements AwsIotCredentialsProvider {
  @override
  Future<AwsIotCredentials> getCredentials() async {
    final session = await Amplify.Auth.fetchAuthSession() as CognitoAuthSession;
    final c = session.credentialsResult.value;
    return AwsIotCredentials(
      accessKeyId: c.accessKeyId,
      secretAccessKey: c.secretAccessKey,
      sessionToken: c.sessionToken,
      expiration: c.expiration,
    );
  }
}

Device Shadow is one call per operation:

final shadow = client.shadow('lamp-1');

final doc = await shadow.get();
print('v${doc.version}: reported ${doc.reported}');

await shadow.update(desired: {'brightness': 80});
shadow.delta.listen((d) => print('to apply: ${d.state}'));

What it does for you:

Fetches credentials and re-signs the URL on every connect and reconnect, with backoff and jitter.
Restores subscriptions after reconnects, with + and # wildcards and reference counting, so two listeners on one topic share a single broker subscription.
Provides Device Shadow get, update and delete with clientToken matching and timeouts, plus delta and documents streams, for classic and named shadows.
Offers a beforeConnect hook, for example to call your attach-policy endpoint.
Throws typed errors from a sealed AwsIotException hierarchy.

It runs on Android, iOS, macOS, Windows, Linux and the web. Before publishing, I tested publish and subscribe, a forced disconnect with reconnect and resubscribe, and Device Shadow against a real AWS IoT broker, on the Dart VM and in Chrome.

The Chrome run was worth it. The shadow client built its request tokens with nextInt(1 << 32). On the web, Dart integers are JavaScript numbers and shifts work on 32 bits, so 1 << 32 is 0, and every shadow request threw. The unit tests passed on the VM. Now CI runs the whole test suite in Chrome too.

Try it, and help improve it

flutter pub add aws_iot_core

The code is on GitHub, and the repository has a runnable example app with an Amplify sign-in screen. Bug reports, fixes and improvements are welcome. The roadmap includes persistent sessions, an offline publish queue, MQTT 5 and X.509 authentication, and any of these is a good first contribution.