Avis de fin de support : le 7 octobre 2026, AWS le support de. AWS IoT Greengrass Version 1 Après le 7 octobre 2026, vous ne pourrez plus accéder aux AWS IoT Greengrass V1 ressources. Pour plus d'informations, rendez-vous sur [Migrer depuis AWS IoT Greengrass Version 1](https://docs.aws.amazon.com/greengrass/v2/developerguide/migrate-from-v1.html).
Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
# StreamManagerClient À utiliser pour travailler avec des flux
Les fonctions Lambda définies par l'utilisateur exécutées sur AWS IoT Greengrass le noyau peuvent utiliser `StreamManagerClient` l'objet du SDK [Core pour créer des flux dans AWS IoT Greengrass le](lambda-functions.md#lambda-sdks) [gestionnaire de flux, puis interagir avec les flux](stream-manager.md). Lorsqu'une fonction Lambda crée un flux, elle définit les AWS Cloud destinations, la hiérarchisation et les autres politiques d'exportation et de conservation des données pour le flux. Pour envoyer des données au gestionnaire de flux, les fonctions Lambda ajoutent les données au flux. Si une destination d'exportation est définie pour le flux, le gestionnaire de flux exporte le flux automatiquement.
**Note**
Généralement, les clients du gestionnaire de flux sont des fonctions Lambda définies par l'utilisateur. Si votre analyse de rentabilisation l'exige, vous pouvez également autoriser les processus non Lambda exécutés sur le cœur de Greengrass (par exemple, un conteneur Docker) à interagir avec le gestionnaire de flux. Pour de plus amples informations, veuillez consulter [Authentification client](stream-manager.md#stream-manager-security-client-authentication).
Les extraits de cette rubrique vous montrent comment les clients appellent des `StreamManagerClient` méthodes pour travailler avec des flux. Pour plus de détails sur l'implémentation des méthodes et de leurs arguments, utilisez les liens vers la référence du SDK répertoriée après chaque extrait. Pour les didacticiels qui incluent une fonction Python Lambda complète, consultez [Exporter des flux de données vers AWS Cloud (console)](stream-manager-console.md) ou. [Exporter des flux de données vers le AWS Cloud (CLI)](stream-manager-cli.md)
Votre fonction Lambda doit être instanciée en `StreamManagerClient` dehors du gestionnaire de fonctions. Si la fonction est instanciée dans le gestionnaire, elle crée un `client` et une connexion au gestionnaire de flux chaque fois qu'elle est appelée.
**Note**
Si vous effectuez une instanciation `StreamManagerClient` dans le gestionnaire, vous devez appeler explicitement la méthode `close()` lorsque le `client` termine son travail. Sinon, le `client` maintient la connexion ouverte et un autre thread actif jusqu'à ce que le script se termine.
`StreamManagerClient` prend en charge les opérations suivantes :
+ [Créer un flux de messages](#streammanagerclient-create-message-stream)
+ [Ajouter un message](#streammanagerclient-append-message)
+ [Lire des messages](#streammanagerclient-read-messages)
+ [Afficher la liste des flux](#streammanagerclient-list-streams)
+ [Décrire le flux de messages](#streammanagerclient-describe-message-stream)
+ [Mettre à jour le flux de messages](#streammanagerclient-update-message-stream)
+ [Supprimer le flux de messages](#streammanagerclient-delete-message-stream)
## Créer un flux de messages
Pour créer un flux, une fonction Lambda définie par l'utilisateur appelle la méthode create et transmet un objet. `MessageStreamDefinition` Cet objet spécifie le nom unique du flux et définit comment le gestionnaire de flux doit gérer les nouvelles données lorsque la taille maximale du flux est atteinte. Vous pouvez utiliser `MessageStreamDefinition` et ses types de données (tels que `ExportDefinition`, `StrategyOnFull` et `Persistence`) pour définir d'autres propriétés de flux. Il s’agit des licences suivantes :
+ La cible AWS IoT Analytics, Kinesis Data Streams et AWS IoT SiteWise les destinations Amazon S3 pour les exportations automatiques. Pour de plus amples informations, veuillez consulter [Exporter les configurations pour les AWS Cloud destinations prises en charge](stream-export-configurations.md).
+ Exportez la priorité. Le gestionnaire de flux exporte les flux de priorité supérieure avant les flux de priorité inférieure.
+ Taille de lot et intervalle de lot maximaux pour AWS IoT Analytics Kinesis Data Streams AWS IoT SiteWise et les destinations. Le gestionnaire de flux exporte les messages lorsque l'une ou l'autre des conditions est remplie.
+ Time-to-live (TTL). Temps nécessaire pour garantir que les données du flux sont disponibles pour le traitement. Vous devez vous assurer que les données peuvent être consommées pendant cette période. Il ne s'agit pas d'une stratégie de suppression. Les données peuvent ne pas être supprimées immédiatement après la période de TTL.
+ Persistance des flux. Choisissez d'enregistrer les flux dans le système de fichiers afin de conserver les données lors des redémarrages du noyau ou d'enregistrer les flux en mémoire.
+ Numéro de séquence de départ. Spécifiez le numéro de séquence du message à utiliser comme message de départ lors de l'exportation.
Pour plus d'informations`MessageStreamDefinition`, consultez la référence du SDK pour votre langue cible :
+ [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)dans le SDK Java
+ [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)dans le SDK Node.js
+ [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.MessageStreamDefinition)dans le SDK Python
**Note**
`StreamManagerClient`fournit également une destination cible que vous pouvez utiliser pour exporter des flux vers un serveur HTTP. Cette cible n'est destinée qu'à des fins de test. Il n'est pas stable ni pris en charge pour une utilisation dans des environnements de production.
Après la création d'un flux, vos fonctions Lambda peuvent [ajouter des messages](#streammanagerclient-append-message) au flux pour envoyer des données à exporter et [lire les messages](#streammanagerclient-append-message) du flux pour un traitement local. Le nombre de flux que vous créez dépend de vos capacités matérielles et de votre analyse de rentabilisation. L'une des stratégies consiste à créer un flux pour chaque canal cible dans AWS IoT Analytics le flux de données Kinesis, bien que vous puissiez définir plusieurs cibles pour un flux. Un flux a un cycle de vie durable.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.10.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.5.0 \$1 Java : 1.4.0 \$1 Node.js : 1.6.0
**Note**
La création de flux avec une destination d'exportation AWS IoT SiteWise ou Amazon S3 répond aux exigences suivantes :
Version minimale AWS IoT Greengrass du noyau : 1.11.0
Version minimale AWS IoT Greengrass du SDK de base : Python : 1.6.0 \$1 Java : 1.5.0 \$1 Node.js : 1.7.0
### Exemples
L'extrait de code suivant crée un flux nommé `StreamName`. Il définit les propriétés du flux dans les types de données `MessageStreamDefinition` et les types de données subordonnés.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
client.create_message_stream(MessageStreamDefinition(
name="StreamName", # Required.
max_size=268435456, # Default is 256 MB.
stream_segment_size=16777216, # Default is 16 MB.
time_to_live_millis=None, # By default, no TTL is enabled.
strategy_on_full=StrategyOnFull.OverwriteOldestData, # Required.
persistence=Persistence.File, # Default is File.
flush_on_write=False, # Default is false.
export_definition=ExportDefinition( # Optional. Choose where/how the stream is exported to the AWS Cloud.
kinesis=None,
iot_analytics=None,
iot_sitewise=None,
s3_task_executor=None
)
))
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
Référence du SDK Python : [create\$1message\$1stream](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.create_message_stream) \$1 [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.MessageStreamDefinition)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
client.createMessageStream(
new MessageStreamDefinition()
.withName("StreamName") // Required.
.withMaxSize(268435456L) // Default is 256 MB.
.withStreamSegmentSize(16777216L) // Default is 16 MB.
.withTimeToLiveMillis(null) // By default, no TTL is enabled.
.withStrategyOnFull(StrategyOnFull.OverwriteOldestData) // Required.
.withPersistence(Persistence.File) // Default is File.
.withFlushOnWrite(false) // Default is false.
.withExportDefinition( // Optional. Choose where/how the stream is exported to the AWS Cloud.
new ExportDefinition()
.withKinesis(null)
.withIotAnalytics(null)
.withIotSitewise(null)
.withS3TaskExecutor(null)
)
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Référence du SDK Java : \$1 [createMessageStream[MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#createMessageStream-com.amazonaws.greengrass.streammanager.model.MessageStreamDefinition-)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
await client.createMessageStream(
new MessageStreamDefinition()
.withName("StreamName") // Required.
.withMaxSize(268435456) // Default is 256 MB.
.withStreamSegmentSize(16777216) // Default is 16 MB.
.withTimeToLiveMillis(null) // By default, no TTL is enabled.
.withStrategyOnFull(StrategyOnFull.OverwriteOldestData) // Required.
.withPersistence(Persistence.File) // Default is File.
.withFlushOnWrite(false) // Default is false.
.withExportDefinition( // Optional. Choose where/how the stream is exported to the AWS Cloud.
new ExportDefinition()
.withKinesis(null)
.withIotAnalytics(null)
.withIotSitewise(null)
.withS3TaskExecutor(null)
)
);
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK Node.js : \$1 [createMessageStream[MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#createMessageStream)
------
Pour plus d'informations sur la configuration des destinations d'exportation, consultez[Exporter les configurations pour les AWS Cloud destinations prises en charge](stream-export-configurations.md).
## Ajouter un message
Pour envoyer des données au gestionnaire de flux à des fins d'exportation, vos fonctions Lambda ajoutent les données au flux cible. La destination d'exportation détermine le type de données à transmettre à cette méthode.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.10.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.5.0 \$1 Java : 1.4.0 \$1 Node.js : 1.6.0
**Note**
L'ajout de messages à une destination d'exportation AWS IoT SiteWise ou à une destination d'exportation Amazon S3 répond aux exigences suivantes :
Version minimale AWS IoT Greengrass du noyau : 1.11.0
Version minimale AWS IoT Greengrass du SDK de base : Python : 1.6.0 \$1 Java : 1.5.0 \$1 Node.js : 1.7.0
### Exemples
#### AWS IoT Analytics ou destinations d'exportation Kinesis Data Streams
L'extrait de code suivant ajoute un message au flux nommé `StreamName`. Pour AWS IoT Analytics nos destinations Kinesis Data Streams, vos fonctions Lambda ajoutent un blob de données.
Cet extrait répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.10.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.5.0 \$1 Java : 1.4.0 \$1 Node.js : 1.6.0
------
#### [ Python ]
```
client = StreamManagerClient()
try:
sequence_number = client.append_message(stream_name="StreamName", data=b'Arbitrary bytes data')
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
[Référence du SDK Python : append\$1message](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.append_message)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
long sequenceNumber = client.appendMessage("StreamName", "Arbitrary byte array".getBytes());
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
[Référence du SDK Java : AppendMessage](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const sequenceNumber = await client.appendMessage("StreamName", Buffer.from("Arbitrary byte array"));
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
[Référence du SDK Node.js : AppendMessage](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage)
------
#### AWS IoT SiteWise destinations d'exportation
L'extrait de code suivant ajoute un message au flux nommé `StreamName`. Pour les AWS IoT SiteWise destinations, vos fonctions Lambda ajoutent un objet sérialisé. `PutAssetPropertyValueEntry` Pour de plus amples informations, veuillez consulter [Exporter vers AWS IoT SiteWise](stream-export-configurations.md#export-streams-to-sitewise).
**Note**
Lorsque vous envoyez des données à AWS IoT SiteWise, celles-ci doivent répondre aux exigences de l'`BatchPutAssetPropertyValue`action. Pour plus d’informations, consultez [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) dans la *Référence d’API AWS IoT SiteWise *.
Cet extrait répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.11.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.6.0 \$1 Java : 1.5.0 \$1 Node.js : 1.7.0
------
#### [ Python ]
```
client = StreamManagerClient()
try:
# SiteWise requires unique timestamps in all messages. Add some randomness to time and offset.
# Note: To create a new asset property data, you should use the classes defined in the
# greengrasssdk.stream_manager module.
time_in_nanos = TimeInNanos(
time_in_seconds=calendar.timegm(time.gmtime()) - random.randint(0, 60), offset_in_nanos=random.randint(0, 10000)
)
variant = Variant(double_value=random.random())
asset = [AssetPropertyValue(value=variant, quality=Quality.GOOD, timestamp=time_in_nanos)]
putAssetPropertyValueEntry = PutAssetPropertyValueEntry(entry_id=str(uuid.uuid4()), property_alias="PropertyAlias", property_values=asset)
sequence_number = client.append_message(stream_name="StreamName", data=Util.validate_and_serialize_to_json_bytes(putAssetPropertyValueEntry))
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
Référence du SDK Python : [append\$1message \$1 [PutAssetPropertyValueEntry](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.PutAssetPropertyValueEntry)](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.append_message)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
Random rand = new Random();
// Note: To create a new asset property data, you should use the classes defined in the
// com.amazonaws.greengrass.streammanager.model.sitewise package.
List entries = new ArrayList<>() ;
// IoTSiteWise requires unique timestamps in all messages. Add some randomness to time and offset.
final int maxTimeRandomness = 60;
final int maxOffsetRandomness = 10000;
double randomValue = rand.nextDouble();
TimeInNanos timestamp = new TimeInNanos()
.withTimeInSeconds(Instant.now().getEpochSecond() - rand.nextInt(maxTimeRandomness))
.withOffsetInNanos((long) (rand.nextInt(maxOffsetRandomness)));
AssetPropertyValue entry = new AssetPropertyValue()
.withValue(new Variant().withDoubleValue(randomValue))
.withQuality(Quality.GOOD)
.withTimestamp(timestamp);
entries.add(entry);
PutAssetPropertyValueEntry putAssetPropertyValueEntry = new PutAssetPropertyValueEntry()
.withEntryId(UUID.randomUUID().toString())
.withPropertyAlias("PropertyAlias")
.withPropertyValues(entries);
long sequenceNumber = client.appendMessage("StreamName", ValidateAndSerialize.validateAndSerializeToJsonBytes(putAssetPropertyValueEntry));
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Référence du SDK Java : [AppendMessage \$1 [PutAssetPropertyValueEntry](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/sitewise/PutAssetPropertyValueEntry.html)](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const maxTimeRandomness = 60;
const maxOffsetRandomness = 10000;
const randomValue = Math.random();
// Note: To create a new asset property data, you should use the classes defined in the
// aws-greengrass-core-sdk StreamManager module.
const timestamp = new TimeInNanos()
.withTimeInSeconds(Math.round(Date.now() / 1000) - Math.floor(Math.random() - maxTimeRandomness))
.withOffsetInNanos(Math.floor(Math.random() * maxOffsetRandomness));
const entry = new AssetPropertyValue()
.withValue(new Variant().withDoubleValue(randomValue))
.withQuality(Quality.GOOD)
.withTimestamp(timestamp);
const putAssetPropertyValueEntry = new PutAssetPropertyValueEntry()
.withEntryId(`${ENTRY_ID_PREFIX}${i}`)
.withPropertyAlias("PropertyAlias")
.withPropertyValues([entry]);
const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(putAssetPropertyValueEntry));
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK Node.js : [AppendMessage \$1 [PutAssetPropertyValueEntry](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.PutAssetPropertyValueEntry.html)](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage)
------
#### Destinations d'exportation Amazon S3
L'extrait suivant ajoute une tâche d'exportation au flux nommé. `StreamName` Pour les destinations Amazon S3, vos fonctions Lambda ajoutent un `S3ExportTaskDefinition` objet sérialisé contenant des informations sur le fichier d'entrée source et l'objet Amazon S3 cible. Si l'objet spécifié n'existe pas, Stream Manager le crée pour vous. Pour de plus amples informations, veuillez consulter [Exportation vers Amazon S3](stream-export-configurations.md#export-streams-to-s3).
Cet extrait répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.11.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.6.0 \$1 Java : 1.5.0 \$1 Node.js : 1.7.0
------
#### [ Python ]
```
client = StreamManagerClient()
try:
# Append an Amazon S3 Task definition and print the sequence number.
s3_export_task_definition = S3ExportTaskDefinition(input_url="URLToFile", bucket="BucketName", key="KeyName")
sequence_number = client.append_message(stream_name="StreamName", data=Util.validate_and_serialize_to_json_bytes(s3_export_task_definition))
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
[Référence du SDK Python : [append\$1message \$1 S3](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.append_message) ExportTaskDefinition](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.S3ExportTaskDefinition)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
// Append an Amazon S3 export task definition and print the sequence number.
S3ExportTaskDefinition s3ExportTaskDefinition = new S3ExportTaskDefinition()
.withBucket("BucketName")
.withKey("KeyName")
.withInputUrl("URLToFile");
long sequenceNumber = client.appendMessage("StreamName", ValidateAndSerialize.validateAndSerializeToJsonBytes(s3ExportTaskDefinition));
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
[Référence du SDK Java : [AppendMessage \$1](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) S3 ExportTaskDefinition](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/S3ExportTaskDefinition.html)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
// Append an Amazon S3 export task definition and print the sequence number.
const taskDefinition = new S3ExportTaskDefinition()
.withBucket("BucketName")
.withKey("KeyName")
.withInputUrl("URLToFile");
const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(taskDefinition)));
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
[Référence du SDK Node.js : [AppendMessage \$1](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) S3 ExportTaskDefinition](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskDefinition.html)
------
## Lire des messages
Lisez les messages d'un stream.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.10.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.5.0 \$1 Java : 1.4.0 \$1 Node.js : 1.6.0
### Exemples
L'extrait de code suivant lit les messages du flux nommé `StreamName`. La méthode read utilise un objet `ReadMessagesOptions` facultatif qui spécifie le numéro de séquence à partir duquel commencer la lecture, les nombres minimum et maximum à lire et un délai d'expiration pour la lecture des messages.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
message_list = client.read_messages(
stream_name="StreamName",
# By default, if no options are specified, it tries to read one message from the beginning of the stream.
options=ReadMessagesOptions(
desired_start_sequence_number=100,
# Try to read from sequence number 100 or greater. By default, this is 0.
min_message_count=10,
# Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is raised. By default, this is 1.
max_message_count=100, # Accept up to 100 messages. By default this is 1.
read_timeout_millis=5000
# Try to wait at most 5 seconds for the min_messsage_count to be fulfilled. By default, this is 0, which immediately returns the messages or an exception.
)
)
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
Référence du SDK Python : [read\$1messages \$1 [ReadMessagesOptions](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.ReadMessagesOptions)](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.read_messages)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
List messages = client.readMessages("StreamName",
// By default, if no options are specified, it tries to read one message from the beginning of the stream.
new ReadMessagesOptions()
// Try to read from sequence number 100 or greater. By default this is 0.
.withDesiredStartSequenceNumber(100L)
// Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is raised. By default, this is 1.
.withMinMessageCount(10L)
// Accept up to 100 messages. By default this is 1.
.withMaxMessageCount(100L)
// Try to wait at most 5 seconds for the min_messsage_count to be fulfilled. By default, this is 0, which immediately returns the messages or an exception.
.withReadTimeoutMillis(Duration.ofSeconds(5L).toMillis())
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Référence du SDK Java : [ReadMessages \$1 [ReadMessagesOptions](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/ReadMessagesOptions.html)](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const messages = await client.readMessages("StreamName",
// By default, if no options are specified, it tries to read one message from the beginning of the stream.
new ReadMessagesOptions()
// Try to read from sequence number 100 or greater. By default this is 0.
.withDesiredStartSequenceNumber(100)
// Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is thrown. By default, this is 1.
.withMinMessageCount(10)
// Accept up to 100 messages. By default this is 1.
.withMaxMessageCount(100)
// Try to wait at most 5 seconds for the minMessageCount to be fulfilled. By default, this is 0, which immediately returns the messages or an exception.
.withReadTimeoutMillis(5 * 1000)
);
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK Node.js : [ReadMessages \$1 [ReadMessagesOptions](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.ReadMessagesOptions.html)](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages)
------
## Afficher la liste des flux
Obtenez la liste des flux dans le gestionnaire de flux.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.10.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.5.0 \$1 Java : 1.4.0 \$1 Node.js : 1.6.0
### Exemples
L'extrait de code suivant récupère une liste des flux (par nom) dans le gestionnaire de flux.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
stream_names = client.list_streams()
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
[Référence du SDK Python : list\$1streams](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.list_streams)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
List streamNames = client.listStreams();
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
[Référence du SDK Java : ListStreams](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#listStreams--)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const streams = await client.listStreams();
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
[Référence du SDK Node.js : ListStreams](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#listStreams)
------
## Décrire le flux de messages
Obtenez les métadonnées relatives à un flux, notamment sa définition, sa taille et son statut d'exportation.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.10.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.5.0 \$1 Java : 1.4.0 \$1 Node.js : 1.6.0
### Exemples
L'extrait de code suivant récupère des métadonnées sur le flux nommé `StreamName`, en particulier la définition, la taille et les statuts d'exportation du flux.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
stream_description = client.describe_message_stream(stream_name="StreamName")
if stream_description.export_statuses[0].error_message:
# The last export of export destination 0 failed with some error
# Here is the last sequence number that was successfully exported
stream_description.export_statuses[0].last_exported_sequence_number
if (stream_description.storage_status.newest_sequence_number >
stream_description.export_statuses[0].last_exported_sequence_number):
pass
# The end of the stream is ahead of the last exported sequence number
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
[Référence du SDK Python : describe\$1message\$1stream](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.describe_message_stream)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
MessageStreamInfo description = client.describeMessageStream("StreamName");
String lastErrorMessage = description.getExportStatuses().get(0).getErrorMessage();
if (lastErrorMessage != null && !lastErrorMessage.equals("")) {
// The last export of export destination 0 failed with some error.
// Here is the last sequence number that was successfully exported.
description.getExportStatuses().get(0).getLastExportedSequenceNumber();
}
if (description.getStorageStatus().getNewestSequenceNumber() >
description.getExportStatuses().get(0).getLastExportedSequenceNumber()) {
// The end of the stream is ahead of the last exported sequence number.
}
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Référence du SDK Java : [describeMessageStream](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#describeMessageStream-java.lang.String-)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const description = await client.describeMessageStream("StreamName");
const lastErrorMessage = description.exportStatuses[0].errorMessage;
if (lastErrorMessage) {
// The last export of export destination 0 failed with some error.
// Here is the last sequence number that was successfully exported.
description.exportStatuses[0].lastExportedSequenceNumber;
}
if (description.storageStatus.newestSequenceNumber >
description.exportStatuses[0].lastExportedSequenceNumber) {
// The end of the stream is ahead of the last exported sequence number.
}
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK Node.js : [describeMessageStream](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#describeMessageStream)
------
## Mettre à jour le flux de messages
Mettez à jour les propriétés d'un flux existant. Vous souhaiterez peut-être mettre à jour un flux si vos exigences changent après sa création. Par exemple :
+ Ajoutez une nouvelle [configuration d'exportation](stream-export-configurations.md) pour une AWS Cloud destination.
+ Augmentez la taille maximale d'un flux pour modifier la façon dont les données sont exportées ou conservées. Par exemple, la taille du flux associée à votre stratégie en matière de paramètres complets peut entraîner la suppression ou le rejet de données avant que le gestionnaire de flux ne puisse les traiter.
+ Interrompez et reprenez les exportations, par exemple, si les tâches d'exportation sont longues et que vous souhaitez rationner vos données de téléchargement.
Vos fonctions Lambda suivent ce processus de haut niveau pour mettre à jour un flux :
1. [Obtenez la description du stream.](#streammanagerclient-describe-message-stream)
1. Mettez à jour les propriétés cibles sur les objets correspondants `MessageStreamDefinition` et subordonnés.
1. Transmettez la mise à jour`MessageStreamDefinition`. Assurez-vous d'inclure les définitions d'objets complètes pour le flux mis à jour. Les propriétés non définies reprennent leurs valeurs par défaut.
Vous pouvez spécifier le numéro de séquence du message à utiliser comme message de départ lors de l'exportation.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.11.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.6.0 \$1 Java : 1.5.0 \$1 Node.js : 1.7.0
### Exemples
L'extrait suivant met à jour le flux nommé. `StreamName` Elle met à jour plusieurs propriétés d'un flux exporté vers Kinesis Data Streams.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
message_stream_info = client.describe_message_stream(STREAM_NAME)
message_stream_info.definition.max_size=536870912
message_stream_info.definition.stream_segment_size=33554432
message_stream_info.definition.time_to_live_millis=3600000
message_stream_info.definition.strategy_on_full=StrategyOnFull.RejectNewData
message_stream_info.definition.persistence=Persistence.Memory
message_stream_info.definition.flush_on_write=False
message_stream_info.definition.export_definition.kinesis=
[KinesisConfig(
# Updating Export definition to add a Kinesis Stream configuration.
identifier=str(uuid.uuid4()), kinesis_stream_name=str(uuid.uuid4()))]
client.update_message_stream(message_stream_info.definition)
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
Référence du SDK Python : \$1 [updateMessageStream[MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.MessageStreamDefinition)](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.update_message_stream)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
MessageStreamInfo messageStreamInfo = client.describeMessageStream(STREAM_NAME);
// Update the message stream with new values.
client.updateMessageStream(
messageStreamInfo.getDefinition()
.withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data.
// Max Size update should be greater than initial Max Size defined in Create Message Stream request
.withMaxSize(536870912L) // Update Max Size to 512 MB.
.withStreamSegmentSize(33554432L) // Update Segment Size to 32 MB.
.withFlushOnWrite(true) // Update flush on write to true.
.withPersistence(Persistence.Memory) // Update the persistence to Memory.
.withTimeToLiveMillis(3600000L) // Update TTL to 1 hour.
.withExportDefinition(
// Optional. Choose where/how the stream is exported to the AWS Cloud.
messageStreamInfo.getDefinition().getExportDefinition().
// Updating Export definition to add a Kinesis Stream configuration.
.withKinesis(new ArrayList() {{
add(new KinesisConfig()
.withIdentifier(EXPORT_IDENTIFIER)
.withKinesisStreamName("test"));
}})
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Référence du SDK Java : [update\$1message\$1stream](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#updateMessageStream-java.lang.String-) \$1 [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const messageStreamInfo = await c.describeMessageStream(STREAM_NAME);
await client.updateMessageStream(
messageStreamInfo.definition
// Max Size update should be greater than initial Max Size defined in Create Message Stream request
.withMaxSize(536870912) // Default is 256 MB. Updating Max Size to 512 MB.
.withStreamSegmentSize(33554432) // Default is 16 MB. Updating Segment Size to 32 MB.
.withTimeToLiveMillis(3600000) // By default, no TTL is enabled. Update TTL to 1 hour.
.withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data.
.withPersistence(Persistence.Memory) // Default is File. Update the persistence to Memory
.withFlushOnWrite(true) // Default is false. Updating to true.
.withExportDefinition(
// Optional. Choose where/how the stream is exported to the AWS Cloud.
messageStreamInfo.definition.exportDefinition
// Updating Export definition to add a Kinesis Stream configuration.
.withKinesis([new KinesisConfig().withIdentifier(uuidv4()).withKinesisStreamName(uuidv4())])
)
);
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK Node.js : \$1 [updateMessageStream[MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#updateMessageStream)
------
### Contraintes de mise à jour des flux
Les contraintes suivantes s'appliquent lors de la mise à jour des flux. Sauf indication contraire dans la liste suivante, les mises à jour prennent effet immédiatement.
+ Vous ne pouvez pas mettre à jour la persistance d'un flux. Pour modifier ce comportement, [supprimez le flux](#streammanagerclient-delete-message-stream) et [créez un flux](#streammanagerclient-create-message-stream) qui définit la nouvelle politique de persistance.
+ Vous pouvez mettre à jour la taille maximale d'un flux uniquement dans les conditions suivantes :
+ La taille maximale doit être supérieure ou égale à la taille actuelle du flux. Pour trouver ces informations, [décrivez le flux](#streammanagerclient-describe-message-stream), puis vérifiez l'état de stockage de l'`MessageStreamInfo`objet renvoyé.
+ La taille maximale doit être supérieure ou égale à la taille du segment du flux.
+ Vous pouvez mettre à jour la taille du segment de flux à une valeur inférieure à la taille maximale du flux. Le paramètre mis à jour s'applique aux nouveaux segments.
+ Les mises à jour de la propriété Time to Live (TTL) s'appliquent aux nouvelles opérations d'ajout. Si vous diminuez cette valeur, le gestionnaire de flux peut également supprimer les segments existants qui dépassent le TTL.
+ Les mises à jour de la stratégie relative à la propriété complète s'appliquent aux nouvelles opérations d'ajout. Si vous définissez une stratégie pour remplacer les données les plus anciennes, le gestionnaire de flux peut également remplacer les segments existants en fonction du nouveau paramètre.
+ Les mises à jour de la propriété flush on write s'appliquent aux nouveaux messages.
+ Les mises à jour des configurations d'exportation s'appliquent aux nouvelles exportations. La demande de mise à jour doit inclure toutes les configurations d'exportation que vous souhaitez prendre en charge. Dans le cas contraire, le gestionnaire de flux les supprime.
+ Lorsque vous mettez à jour une configuration d'exportation, spécifiez l'identifiant de la configuration d'exportation cible.
+ Pour ajouter une configuration d'exportation, spécifiez un identifiant unique pour la nouvelle configuration d'exportation.
+ Pour supprimer une configuration d'exportation, omettez-la.
+ Pour [mettre à jour](#streammanagerclient-update-message-stream) le numéro de séquence de départ d'une configuration d'exportation dans un flux, vous devez spécifier une valeur inférieure au dernier numéro de séquence. Pour trouver ces informations, [décrivez le flux](#streammanagerclient-describe-message-stream), puis vérifiez l'état de stockage de l'`MessageStreamInfo`objet renvoyé.
## Supprimer le flux de messages
Supprime un flux. Lorsque vous supprimez un flux, toutes les données stockées dans le flux sont supprimées du disque.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale AWS IoT Greengrass du noyau : 1.10.0
+ Version minimale AWS IoT Greengrass du SDK de base : Python : 1.5.0 \$1 Java : 1.4.0 \$1 Node.js : 1.6.0
### Exemples
L'extrait de code suivant supprime le flux nommé `StreamName`.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
client.delete_message_stream(stream_name="StreamName")
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
Référence du SDK Python : [deleteMessageStream](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.delete_message_stream)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
client.deleteMessageStream("StreamName");
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
[Référence du SDK Java : delete\$1message\$1stream](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#deleteMessageStream-java.lang.String-)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
await client.deleteMessageStream("StreamName");
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK Node.js : [deleteMessageStream](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#deleteMessageStream)
------
## Consultez aussi
+ [Gérez les flux de données sur le AWS IoT Greengrass cœur](stream-manager.md)
+ [Configuration du gestionnaire de AWS IoT Greengrass flux](configure-stream-manager.md)
+ [Exporter les configurations pour les AWS Cloud destinations prises en charge](stream-export-configurations.md)
+ [Exporter des flux de données vers AWS Cloud (console)](stream-manager-console.md)
+ [Exporter des flux de données vers le AWS Cloud (CLI)](stream-manager-cli.md)
+ `StreamManagerClient`dans la référence du SDK AWS IoT Greengrass Core :
+ [Python](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html)
+ [Java](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html)
+ [Node.js](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html)
# Exporter les configurations pour les AWS Cloud destinations prises en charge
Les fonctions Lambda définies par l'utilisateur sont `StreamManagerClient` utilisées dans AWS IoT Greengrass le SDK Core pour interagir avec le gestionnaire de flux. Lorsqu'une fonction Lambda [crée un flux](work-with-streams.md#streammanagerclient-create-message-stream) ou [met à jour un flux](work-with-streams.md#streammanagerclient-create-message-stream), elle transmet un `MessageStreamDefinition` objet qui représente les propriétés du flux, y compris la définition de l'exportation. L'`ExportDefinition`objet contient les configurations d'exportation définies pour le flux. Le gestionnaire de flux utilise ces configurations d'exportation pour déterminer où et comment exporter le flux.
![\[Schéma du modèle d'objet du type de ExportDefinition propriété.\]](http://docs.aws.amazon.com/fr_fr/greengrass/v1/developerguide/images/stream-manager-exportconfigs.png)
Vous pouvez définir zéro ou plusieurs configurations d'exportation sur un flux, y compris plusieurs configurations d'exportation pour un seul type de destination. Par exemple, vous pouvez exporter un flux vers deux AWS IoT Analytics canaux et un flux de données Kinesis.
En cas d'échec des tentatives d'exportation, le gestionnaire de flux essaie continuellement d'exporter les données à des AWS Cloud intervalles allant jusqu'à cinq minutes. Le nombre de nouvelles tentatives n'est pas limité.
**Note**
`StreamManagerClient`fournit également une destination cible que vous pouvez utiliser pour exporter des flux vers un serveur HTTP. Cette cible n'est destinée qu'à des fins de test. Il n'est pas stable ni pris en charge pour une utilisation dans des environnements de production.
**Topics**
+ [AWS IoT Analytics chaînes](#export-to-iot-analytics)
+ [Flux de données Amazon Kinesis](#export-to-kinesis)
+ [AWS IoT SiteWise propriétés des actifs](#export-to-iot-sitewise)
+ [Objets Amazon S3](#export-to-s3)
Vous êtes responsable de la maintenance de ces AWS Cloud ressources.
## AWS IoT Analytics chaînes
Le gestionnaire de flux prend en charge les exportations automatiques vers AWS IoT Analytics. AWS IoT Analytics vous permet d'effectuer une analyse avancée de vos données afin de prendre des décisions commerciales et d'améliorer les modèles d'apprentissage automatique. Pour plus d'informations, voir [Qu'est-ce que c'est AWS IoT Analytics ?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) dans le *guide de AWS IoT Analytics l'utilisateur*.
Dans le SDK AWS IoT Greengrass principal, vos fonctions Lambda utilisent `IoTAnalyticsConfig` le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :
+ [Io TAnalytics Config](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.IoTAnalyticsConfig) dans le SDK Python
+ [Io TAnalytics Config](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTAnalyticsConfig.html) dans le SDK Java
+ [Io TAnalytics Config](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTAnalyticsConfig.html) dans le SDK Node.js
### Exigences
Cette destination d'exportation répond aux exigences suivantes :
+ Les canaux cibles entrants AWS IoT Analytics doivent appartenir au même groupe Greengrass Compte AWS et Région AWS appartenir à celui-ci.
+ Ils [Rôle de groupe Greengrass](group-role.md) doivent `iotanalytics:BatchPutMessage` autoriser les chaînes cibles. Par exemple :
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"iotanalytics:BatchPutMessage"
],
"Resource": [
"arn:aws:iotanalytics:us-east-1:123456789012:channel/channel_1_name",
"arn:aws:iotanalytics:us-east-1:123456789012:channel/channel_2_name"
]
}
]
}
```
------
Vous pouvez accorder un accès granulaire ou conditionnel aux ressources, par exemple en utilisant un schéma de `*` dénomination générique. Pour plus d'informations, consultez la section [Ajout et suppression de politiques IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dans le Guide de l'*utilisateur IAM*.
### Exporter vers AWS IoT Analytics
Pour créer un flux exporté vers AWS IoT Analytics, vos fonctions Lambda [créent un flux avec une](work-with-streams.md#streammanagerclient-create-message-stream) définition d'exportation qui inclut un ou plusieurs `IoTAnalyticsConfig` objets. Cet objet définit les paramètres d'exportation, tels que le canal cible, la taille du lot, l'intervalle entre les lots et la priorité.
Lorsque vos fonctions Lambda reçoivent des données provenant d'appareils, elles [ajoutent des messages](work-with-streams.md#streammanagerclient-append-message) contenant un blob de données au flux cible.
Le gestionnaire de flux exporte ensuite les données en fonction des paramètres de lot et de la priorité définis dans les configurations d'exportation du flux.
## Flux de données Amazon Kinesis
Le gestionnaire de flux prend en charge les exportations automatiques vers Amazon Kinesis Data Streams. Kinesis Data Streams est couramment utilisé pour agréger de gros volumes de données et les charger dans un entrepôt de données ou un cluster Map-Reduce. Pour plus d'informations, consultez [Qu'est-ce qu'Amazon Kinesis Data Streams ?](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) dans le manuel *Amazon Kinesis Developer Guide*.
Dans le SDK AWS IoT Greengrass principal, vos fonctions Lambda utilisent `KinesisConfig` le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :
+ [KinesisConfig](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.KinesisConfig)dans le SDK Python
+ [KinesisConfig](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/export/KinesisConfig.html)dans le SDK Java
+ [KinesisConfig](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.KinesisConfig.html)dans le SDK Node.js
### Exigences
Cette destination d'exportation répond aux exigences suivantes :
+ Les flux cibles dans Kinesis Data Streams doivent appartenir au Compte AWS même groupe Greengrass Région AWS et appartenir à celui-ci.
+ Ils [Rôle de groupe Greengrass](group-role.md) doivent `kinesis:PutRecords` autoriser le ciblage des flux de données. Par exemple :
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"kinesis:PutRecords"
],
"Resource": [
"arn:aws:kinesis:us-east-1:123456789012:stream/stream_1_name",
"arn:aws:kinesis:us-east-1:123456789012:stream/stream_2_name"
]
}
]
}
```
------
Vous pouvez accorder un accès granulaire ou conditionnel aux ressources, par exemple en utilisant un schéma de `*` dénomination générique. Pour plus d'informations, consultez la section [Ajout et suppression de politiques IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dans le Guide de l'*utilisateur IAM*.
### Exportation vers Kinesis Data Streams
Pour créer un flux exporté vers Kinesis Data Streams, vos [fonctions Lambda créent un flux avec une](work-with-streams.md#streammanagerclient-create-message-stream) définition d'exportation qui inclut un ou plusieurs objets. `KinesisConfig` Cet objet définit les paramètres d'exportation, tels que le flux de données cible, la taille du lot, l'intervalle entre les lots et la priorité.
Lorsque vos fonctions Lambda reçoivent des données provenant d'appareils, elles [ajoutent des messages](work-with-streams.md#streammanagerclient-append-message) contenant un blob de données au flux cible. Le gestionnaire de flux exporte ensuite les données en fonction des paramètres de lot et de la priorité définis dans les configurations d'exportation du flux.
Le gestionnaire de flux génère un UUID unique et aléatoire comme clé de partition pour chaque enregistrement chargé sur Amazon Kinesis.
## AWS IoT SiteWise propriétés des actifs
Le gestionnaire de flux prend en charge les exportations automatiques vers AWS IoT SiteWise. AWS IoT SiteWise vous permet de collecter, d'organiser et d'analyser les données des équipements industriels à grande échelle. Pour plus d'informations, voir [Qu'est-ce que c'est AWS IoT SiteWise ?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) dans le *guide de AWS IoT SiteWise l'utilisateur*.
Dans le SDK AWS IoT Greengrass principal, vos fonctions Lambda utilisent `IoTSiteWiseConfig` le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :
+ [Io TSite WiseConfig](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.IoTSiteWiseConfig) dans le SDK Python
+ [Io TSite WiseConfig](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTSiteWiseConfig.html) dans le SDK Java
+ [Io TSite WiseConfig](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTSiteWiseConfig.html) dans le SDK Node.js
**Note**
AWS fournit également le[SiteWise Connecteur IoT](iot-sitewise-connector.md), qui est une solution prédéfinie que vous pouvez utiliser avec des sources OPC-UA.
### Exigences
Cette destination d'exportation répond aux exigences suivantes :
+ Les propriétés des actifs cibles AWS IoT SiteWise doivent appartenir aux mêmes Compte AWS et à Région AWS celles du groupe Greengrass.
**Note**
Pour la liste des régions prises AWS IoT SiteWise en charge, voir les [AWS IoT SiteWise points de terminaison et les quotas](https://docs.aws.amazon.com/general/latest/gr/iot-sitewise.html#iot-sitewise_region) dans la *référence AWS générale*.
+ Ils [Rôle de groupe Greengrass](group-role.md) doivent `iotsitewise:BatchPutAssetPropertyValue` autoriser le ciblage des propriétés des actifs. L'exemple de politique suivant utilise la clé de `iotsitewise:assetHierarchyPath` condition pour accorder l'accès à une ressource racine cible et à ses enfants. Vous pouvez le supprimer `Condition` de la politique pour autoriser l'accès à tous vos AWS IoT SiteWise actifs ou spécifier ARNs des actifs individuels.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "iotsitewise:BatchPutAssetPropertyValue",
"Resource": "*",
"Condition": {
"StringLike": {
"iotsitewise:assetHierarchyPath": [
"/root node asset ID",
"/root node asset ID/*"
]
}
}
}
]
}
```
------
Vous pouvez accorder un accès granulaire ou conditionnel aux ressources, par exemple en utilisant un schéma de `*` dénomination générique. Pour plus d'informations, consultez la section [Ajout et suppression de politiques IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dans le Guide de l'*utilisateur IAM*.
Pour obtenir des informations de sécurité importantes, consultez la section [ BatchPutAssetPropertyValue autorisation](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/security_iam_service-with-iam.html#security_iam_service-with-iam-id-based-policies-batchputassetpropertyvalue-action) dans le *guide de AWS IoT SiteWise l'utilisateur*.
### Exporter vers AWS IoT SiteWise
Pour créer un flux exporté vers AWS IoT SiteWise, vos fonctions Lambda [créent un flux avec une](work-with-streams.md#streammanagerclient-create-message-stream) définition d'exportation qui inclut un ou plusieurs `IoTSiteWiseConfig` objets. Cet objet définit les paramètres d'exportation, tels que la taille du lot, l'intervalle entre les lots et la priorité.
Lorsque vos fonctions Lambda reçoivent des données sur les propriétés des actifs de la part d'appareils, elles ajoutent des messages contenant ces données au flux cible. Les messages sont des `PutAssetPropertyValueEntry` objets sérialisés en JSON qui contiennent des valeurs de propriété pour une ou plusieurs propriétés d'actifs. Pour plus d'informations, voir [Ajouter un message pour](work-with-streams.md#streammanagerclient-append-message-sitewise) les destinations AWS IoT SiteWise d'exportation.
**Note**
Lorsque vous envoyez des données à AWS IoT SiteWise, celles-ci doivent répondre aux exigences de l'`BatchPutAssetPropertyValue`action. Pour plus d’informations, consultez [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) dans la *Référence d’API AWS IoT SiteWise *.
Le gestionnaire de flux exporte ensuite les données en fonction des paramètres de lot et de la priorité définis dans les configurations d'exportation du flux.
Vous pouvez ajuster les paramètres de votre gestionnaire de flux et la logique de la fonction Lambda pour concevoir votre stratégie d'exportation. Par exemple :
+ Pour les exportations en temps quasi réel, définissez des paramètres de taille de lot et d'intervalle faibles et ajoutez les données au flux dès leur réception.
+ Pour optimiser le traitement par lots, atténuer les contraintes de bande passante ou minimiser les coûts, vos fonctions Lambda peuvent regrouper timestamp-quality-value les points de données (TQV) reçus pour une propriété d'actif unique avant d'ajouter les données au flux. L'une des stratégies consiste à regrouper les entrées pour un maximum de 10 combinaisons propriétés-actifs différentes, ou alias de propriété, dans un seul message au lieu d'envoyer plusieurs entrées pour la même propriété. Cela permet au gestionnaire de flux de rester dans les limites [AWS IoT SiteWise des quotas](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/quotas.html).
## Objets Amazon S3
Le gestionnaire de flux prend en charge les exportations automatiques vers Amazon S3. Vous pouvez utiliser Amazon S3 pour stocker et récupérer de grandes quantités de données. Pour plus d'informations, consultez [Qu'est-ce qu'Amazon S3 ?](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) dans le *guide du développeur Amazon Simple Storage Service*.
Dans le SDK AWS IoT Greengrass principal, vos fonctions Lambda utilisent `S3ExportTaskExecutorConfig` le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :
+ [S3 ExportTaskExecutorConfig](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.S3ExportTaskExecutorConfig) dans le SDK Python
+ [S3 ExportTaskExecutorConfig](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/export/S3ExportTaskExecutorConfig.html) dans le SDK Java
+ [S3 ExportTaskExecutorConfig](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskExecutorConfig.html) dans le SDK Node.js
### Exigences
Cette destination d'exportation répond aux exigences suivantes :
+ Les compartiments Amazon S3 cibles doivent appartenir au même groupe Compte AWS que le groupe Greengrass.
+ Si la [conteneurisation par défaut](lambda-group-config.md#lambda-containerization-groupsettings) du groupe Greengrass est le **conteneur** Greengrass, vous devez définir le paramètre [STREAM\$1MANAGER\$1READ\$1ONLY\$1DIRS pour utiliser un répertoire](configure-stream-manager.md#stream-manager-read-only-directories) de fichiers d'entrée situé sous le système de fichiers racine ou non. `/tmp`
+ Si une fonction Lambda exécutée en mode **conteneur Greengrass** écrit des fichiers d'entrée dans le répertoire des fichiers d'entrée, vous devez créer une ressource de volume locale pour le répertoire et monter le répertoire dans le conteneur avec des autorisations d'écriture. Cela garantit que les fichiers sont écrits dans le système de fichiers racine et visibles en dehors du conteneur. Pour de plus amples informations, veuillez consulter [Accédez aux ressources locales grâce aux fonctions et connecteurs Lambda](access-local-resources.md).
+ Ils [Rôle de groupe Greengrass](group-role.md) doivent accorder les autorisations suivantes aux compartiments cibles. Par exemple :
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"s3:PutObject",
"s3:AbortMultipartUpload",
"s3:ListMultipartUploadParts"
],
"Resource": [
"arn:aws:s3:::bucket-1-name/*",
"arn:aws:s3:::bucket-2-name/*"
]
}
]
}
```
------
Vous pouvez accorder un accès granulaire ou conditionnel aux ressources, par exemple en utilisant un schéma de `*` dénomination générique. Pour plus d'informations, consultez la section [Ajout et suppression de politiques IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dans le Guide de l'*utilisateur IAM*.
### Exportation vers Amazon S3
Pour créer un flux exporté vers Amazon S3, vos fonctions Lambda utilisent l'`S3ExportTaskExecutorConfig`objet pour configurer la politique d'exportation. La politique définit les paramètres d'exportation, tels que le seuil et la priorité du téléchargement en plusieurs parties. Pour les exportations Amazon S3, le gestionnaire de flux télécharge les données qu'il lit à partir de fichiers locaux sur l'appareil principal. Pour lancer un téléchargement, vos fonctions Lambda ajoutent une tâche d'exportation au flux cible. La tâche d'exportation contient des informations sur le fichier d'entrée et l'objet Amazon S3 cible. Le gestionnaire de flux exécute les tâches dans l'ordre dans lequel elles sont ajoutées au flux.
**Note**
Le compartiment cible doit déjà exister dans votre Compte AWS. Si aucun objet correspondant à la clé spécifiée n'existe, le gestionnaire de flux le crée pour vous.
Ce flux de travail de haut niveau est illustré dans le schéma suivant.
![\[Schéma du flux de travail du gestionnaire de flux pour les exportations Amazon S3.\]](http://docs.aws.amazon.com/fr_fr/greengrass/v1/developerguide/images/stream-manager-s3.png)
Le gestionnaire de flux utilise la propriété de seuil de téléchargement en plusieurs parties, le paramètre de [taille de pièce minimale](configure-stream-manager.md#stream-manager-minimum-part-size) et la taille du fichier d'entrée pour déterminer le mode de téléchargement des données. Le seuil de téléchargement partitionné doit être supérieur ou égal à la taille de pièce minimale. Si vous souhaitez télécharger des données en parallèle, vous pouvez créer plusieurs flux.
Les clés qui spécifient vos objets Amazon S3 cibles peuvent inclure des DateTimeFormatter chaînes [Java](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html) valides dans les `!{timestamp:value}` espaces réservés. Vous pouvez utiliser ces espaces réservés d'horodatage pour partitionner les données dans Amazon S3 en fonction de l'heure à laquelle les données du fichier d'entrée ont été téléchargées. Par exemple, le nom de clé suivant correspond à une valeur telle que`my-key/2020/12/31/data.txt`.
```
my-key/!{timestamp:YYYY}/!{timestamp:MM}/!{timestamp:dd}/data.txt
```
**Note**
Si vous souhaitez surveiller l'état d'exportation d'un flux, créez d'abord un flux d'état, puis configurez le flux d'exportation pour l'utiliser. Pour de plus amples informations, veuillez consulter [Surveiller les tâches d'exportation](#monitor-export-status-s3).
#### Gérer les données d'entrée
Vous pouvez créer du code que les applications IoT utilisent pour gérer le cycle de vie des données d'entrée. L'exemple de flux de travail suivant montre comment vous pouvez utiliser les fonctions Lambda pour gérer ces données.
1. Un processus local reçoit des données provenant d'appareils ou de périphériques, puis écrit les données dans des fichiers situés dans un répertoire du périphérique principal. Il s'agit des fichiers d'entrée pour le gestionnaire de flux.
**Note**
Pour déterminer si vous devez configurer l'accès au répertoire du fichier d'entrée, consultez le paramètre [STREAM\$1MANAGER\$1READ\$1ONLY\$1DIRS](configure-stream-manager.md#stream-manager-read-only-directories).
Le processus dans lequel le gestionnaire de flux s'exécute hérite de toutes les autorisations du système de fichiers associées à l'[identité d'accès par défaut](lambda-group-config.md#lambda-access-identity-groupsettings) du groupe. Le gestionnaire de flux doit être autorisé à accéder aux fichiers d'entrée. Vous pouvez utiliser la `chmod(1)` commande pour modifier l'autorisation des fichiers, si nécessaire.
1. Une fonction Lambda analyse le répertoire et [ajoute une tâche d'exportation](work-with-streams.md#streammanagerclient-append-message-export-task) au flux cible lorsqu'un nouveau fichier est créé. La tâche est un `S3ExportTaskDefinition` objet sérialisé en JSON qui spécifie l'URL du fichier d'entrée, le compartiment et la clé Amazon S3 cibles, ainsi que les métadonnées utilisateur facultatives.
1. Le gestionnaire de flux lit le fichier d'entrée et exporte les données vers Amazon S3 dans l'ordre des tâches ajoutées. Le compartiment cible doit déjà exister dans votre Compte AWS. Si aucun objet correspondant à la clé spécifiée n'existe, le gestionnaire de flux le crée pour vous.
1. La fonction Lambda [lit les messages](work-with-streams.md#streammanagerclient-read-messages) d'un flux d'état pour surveiller le statut de l'exportation. Une fois les tâches d'exportation terminées, la fonction Lambda peut supprimer les fichiers d'entrée correspondants. Pour de plus amples informations, veuillez consulter [Surveiller les tâches d'exportation](#monitor-export-status-s3).
### Surveiller les tâches d'exportation
Vous pouvez créer du code que les applications IoT utilisent pour surveiller le statut de vos exportations Amazon S3. Vos fonctions Lambda doivent créer un flux d'état, puis configurer le flux d'exportation pour écrire des mises à jour de statut dans le flux d'état. Un seul flux de statut peut recevoir des mises à jour de statut provenant de plusieurs flux exportés vers Amazon S3.
[Créez d'abord un flux](work-with-streams.md#streammanagerclient-create-message-stream) à utiliser comme flux d'état. Vous pouvez configurer la taille et les politiques de rétention du flux afin de contrôler la durée de vie des messages d'état. Par exemple :
+ Définissez `Persistence` cette `Memory` option si vous ne souhaitez pas enregistrer les messages d'état.
+ Réglé `StrategyOnFull` sur pour `OverwriteOldestData` que les nouveaux messages d'état ne soient pas perdus.
Créez ou mettez à jour le flux d'exportation pour utiliser le flux d'état. Spécifiquement, définissez la propriété de configuration d'état de la configuration `S3ExportTaskExecutorConfig` d'exportation du flux. Cela indique au gestionnaire de flux d'écrire des messages d'état concernant les tâches d'exportation dans le flux d'état. Dans l'`StatusConfig`objet, spécifiez le nom du flux d'état et le niveau de verbosité. Les valeurs prises en charge suivantes vont de la moins détaillée (`ERROR`) à la plus détaillée (). `TRACE` La valeur par défaut est `INFO`.
+ `ERROR`
+ `WARN`
+ `INFO`
+ `DEBUG`
+ `TRACE`
L'exemple de flux de travail suivant montre comment les fonctions Lambda peuvent utiliser un flux d'état pour surveiller le statut des exportations.
1. Comme décrit dans le flux de travail précédent, une fonction Lambda [ajoute une tâche d'exportation](work-with-streams.md#streammanagerclient-append-message-export-task) à un flux configuré pour écrire des messages d'état concernant les tâches d'exportation dans un flux de statut. L'opération d'ajout renvoie un numéro de séquence qui représente l'ID de tâche.
1. Une fonction Lambda [lit les messages](work-with-streams.md#streammanagerclient-read-messages) de manière séquentielle à partir du flux d'état, puis filtre les messages en fonction du nom du flux et de l'ID de tâche ou en fonction d'une propriété de tâche d'exportation du contexte du message. Par exemple, la fonction Lambda peut filtrer en fonction de l'URL du fichier d'entrée de la tâche d'exportation, qui est représentée par l'`S3ExportTaskDefinition`objet dans le contexte du message.
Les codes d'état suivants indiquent qu'une tâche d'exportation est terminée :
+ `Success`. Le téléchargement a été effectué avec succès.
+ `Failure`. Le gestionnaire de flux a rencontré une erreur. Par exemple, le bucket spécifié n'existe pas. Une fois le problème résolu, vous pouvez à nouveau ajouter la tâche d'exportation au flux.
+ `Canceled`. La tâche a été abandonnée car le flux ou la définition d'exportation a été supprimé, ou parce que la période time-to-live (TTL) de la tâche a expiré.
**Note**
La tâche peut également avoir le statut `InProgress` ou`Warning`. Le gestionnaire de flux émet des avertissements lorsqu'un événement renvoie une erreur qui n'affecte pas l'exécution de la tâche. Par exemple, l'échec du nettoyage d'un téléchargement partiel interrompu renvoie un avertissement.
1. Une fois les tâches d'exportation terminées, la fonction Lambda peut supprimer les fichiers d'entrée correspondants.
L'exemple suivant montre comment une fonction Lambda peut lire et traiter les messages d'état.
------
#### [ Python ]
```
import time
from greengrasssdk.stream_manager import (
ReadMessagesOptions,
Status,
StatusConfig,
StatusLevel,
StatusMessage,
StreamManagerClient,
)
from greengrasssdk.stream_manager.util import Util
client = StreamManagerClient()
try:
# Read the statuses from the export status stream
is_file_uploaded_to_s3 = False
while not is_file_uploaded_to_s3:
try:
messages_list = client.read_messages(
"StatusStreamName", ReadMessagesOptions(min_message_count=1, read_timeout_millis=1000)
)
for message in messages_list:
# Deserialize the status message first.
status_message = Util.deserialize_json_bytes_to_obj(message.payload, StatusMessage)
# Check the status of the status message. If the status is "Success",
# the file was successfully uploaded to S3.
# If the status was either "Failure" or "Cancelled", the server was unable to upload the file to S3.
# We will print the message for why the upload to S3 failed from the status message.
# If the status was "InProgress", the status indicates that the server has started uploading
# the S3 task.
if status_message.status == Status.Success:
logger.info("Successfully uploaded file at path " + file_url + " to S3.")
is_file_uploaded_to_s3 = True
elif status_message.status == Status.Failure or status_message.status == Status.Canceled:
logger.info(
"Unable to upload file at path " + file_url + " to S3. Message: " + status_message.message
)
is_file_uploaded_to_s3 = True
time.sleep(5)
except StreamManagerException:
logger.exception("Exception while running")
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
Référence du SDK Python : [read\$1messages \$1 [StatusMessage](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.StatusMessage)](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.read_messages)
------
#### [ Java ]
```
import com.amazonaws.greengrass.streammanager.client.StreamManagerClient;
import com.amazonaws.greengrass.streammanager.client.utils.ValidateAndSerialize;
import com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions;
import com.amazonaws.greengrass.streammanager.model.Status;
import com.amazonaws.greengrass.streammanager.model.StatusConfig;
import com.amazonaws.greengrass.streammanager.model.StatusLevel;
import com.amazonaws.greengrass.streammanager.model.StatusMessage;
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
try {
boolean isS3UploadComplete = false;
while (!isS3UploadComplete) {
try {
// Read the statuses from the export status stream
List messages = client.readMessages("StatusStreamName",
new ReadMessagesOptions().withMinMessageCount(1L).withReadTimeoutMillis(1000L));
for (Message message : messages) {
// Deserialize the status message first.
StatusMessage statusMessage = ValidateAndSerialize.deserializeJsonBytesToObj(message.getPayload(), StatusMessage.class);
// Check the status of the status message. If the status is "Success", the file was successfully uploaded to S3.
// If the status was either "Failure" or "Canceled", the server was unable to upload the file to S3.
// We will print the message for why the upload to S3 failed from the status message.
// If the status was "InProgress", the status indicates that the server has started uploading the S3 task.
if (Status.Success.equals(statusMessage.getStatus())) {
System.out.println("Successfully uploaded file at path " + FILE_URL + " to S3.");
isS3UploadComplete = true;
} else if (Status.Failure.equals(statusMessage.getStatus()) || Status.Canceled.equals(statusMessage.getStatus())) {
System.out.println(String.format("Unable to upload file at path %s to S3. Message %s",
statusMessage.getStatusContext().getS3ExportTaskDefinition().getInputUrl(),
statusMessage.getMessage()));
sS3UploadComplete = true;
}
}
} catch (StreamManagerException ignored) {
} finally {
// Sleep for sometime for the S3 upload task to complete before trying to read the status message.
Thread.sleep(5000);
}
} catch (e) {
// Properly handle errors.
}
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Référence du SDK Java : [ReadMessages \$1 [StatusMessage](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/StatusMessage.html)](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-)
------
#### [ Node.js ]
```
const {
StreamManagerClient, ReadMessagesOptions,
Status, StatusConfig, StatusLevel, StatusMessage,
util,
} = require('aws-greengrass-core-sdk').StreamManager;
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
let isS3UploadComplete = false;
while (!isS3UploadComplete) {
try {
// Read the statuses from the export status stream
const messages = await c.readMessages("StatusStreamName",
new ReadMessagesOptions()
.withMinMessageCount(1)
.withReadTimeoutMillis(1000));
messages.forEach((message) => {
// Deserialize the status message first.
const statusMessage = util.deserializeJsonBytesToObj(message.payload, StatusMessage);
// Check the status of the status message. If the status is 'Success', the file was successfully uploaded to S3.
// If the status was either 'Failure' or 'Cancelled', the server was unable to upload the file to S3.
// We will print the message for why the upload to S3 failed from the status message.
// If the status was "InProgress", the status indicates that the server has started uploading the S3 task.
if (statusMessage.status === Status.Success) {
console.log(`Successfully uploaded file at path ${FILE_URL} to S3.`);
isS3UploadComplete = true;
} else if (statusMessage.status === Status.Failure || statusMessage.status === Status.Canceled) {
console.log(`Unable to upload file at path ${FILE_URL} to S3. Message: ${statusMessage.message}`);
isS3UploadComplete = true;
}
});
// Sleep for sometime for the S3 upload task to complete before trying to read the status message.
await new Promise((r) => setTimeout(r, 5000));
} catch (e) {
// Ignored
}
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK Node.js : [ReadMessages \$1 [StatusMessage](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StatusMessage.html)](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages)
------