As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.
# Use StreamManagerClient para trabalhar com streams
Os componentes do Greengrass definidos pelo usuário em execução no dispositivo principal do Greengrass podem usar o objeto `StreamManagerClient` no Stream Manager SDK para criar e interagir com os fluxos no [Gerenciador de fluxos](manage-data-streams.md). Quando um componente cria um fluxo, ele define os Nuvem AWS destinos, a priorização e outras políticas de exportação e retenção de dados para o fluxo. Para enviar os dados para o Gerenciador de fluxos, os componentes anexam os dados ao fluxo. Se um destino de exportação for definido para o fluxo, o gerenciador de fluxo exportará o fluxo automaticamente.
**nota**
Normalmente, os clientes do Gerenciador de fluxos são componentes do Greengrass definidos pelo usuário. Se o seu caso de negócios exigir, você também poderá permitir que os processos que não são de componente e são executados no núcleo do Greengrass (por exemplo, um contêiner do Docker) interajam com o Gerenciador de fluxos. Para obter mais informações, consulte [Autenticação de cliente](manage-data-streams.md#stream-manager-security-client-authentication).
Os snippets neste tópico mostram como os clientes chamam o `StreamManagerClient` para trabalhar com fluxos. Para obter detalhes de implementação sobre os métodos e seus argumentos, use os links para a referência do SDK listada após cada snippet.
Se você usa o Gerenciador de fluxos em uma função do Lambda, ela deve instanciar `StreamManagerClient` fora do manipulador de função. Se instanciado no manipulador, a função cria um `client` e uma conexão para o gerenciador de fluxo sempre que for invocado.
**nota**
Se você instanciar `StreamManagerClient` no manipulador, você deve chamar explicitamente o método `close()` quando o `client` concluir seu trabalho. Caso contrário, o `client` mantém a conexão aberta e outro thread em execução até que o script seja encerrado.
`StreamManagerClient` comporta as operações a seguir:
+ [Criar stream de mensagens](#streammanagerclient-create-message-stream)
+ [Anexar mensagem](#streammanagerclient-append-message)
+ [Ler Mensagens](#streammanagerclient-read-messages)
+ [Listar fluxos](#streammanagerclient-list-streams)
+ [Descrever stream de mensagens](#streammanagerclient-describe-message-stream)
+ [Atualize o fluxo de mensagens](#streammanagerclient-update-message-stream)
+ [Excluir stream de mensagens](#streammanagerclient-delete-message-stream)
## Criar stream de mensagens
Para criar um fluxo, um componente do Greengrass definido pelo usuário chama o método create e especifica um objeto `MessageStreamDefinition`. Esse objeto especifica o nome exclusivo do fluxo e define como o gerenciador de fluxo deve lidar com novos dados quando o tamanho máximo do fluxo for atingido. Você pode usar `MessageStreamDefinition` e os tipos de dados (como `ExportDefinition`, `StrategyOnFull` e `Persistence`) para definir outras propriedades de fluxo. Isso inclui:
+ O destino AWS IoT Analytics, Kinesis Data AWS IoT SiteWise Streams e Amazon S3, destinos para exportações automáticas. Para obter mais informações, consulte [Configurações de exportação para destinos compatíveis Nuvem AWS](stream-export-configurations.md).
+ Prioridade da exportação. O gerenciador de fluxo exporta fluxos de prioridade mais alta antes de fluxos de prioridade mais baixa.
+ Tamanho máximo do lote e intervalo de lote para AWS IoT Analytics Kinesis Data Streams e destinos AWS IoT SiteWise . O gerenciador de fluxo exporta mensagens quando qualquer condição é atendida.
+ Time-to-live (TTL). O tempo necessário para garantir que os dados de fluxo estejam disponíveis para processamento. Você deve certificar-se de que os dados podem ser consumidos nesse período de tempo. Esta não é uma política de exclusão. Os dados podem não ser excluídos imediatamente após o período de TTL.
+ Persistência do fluxo. Selecione salvar fluxos no sistema de arquivos para persistir os dados nas reinicializações do núcleo ou salve os fluxos na memória.
+ Número de sequência inicial. Especifique o número de sequência da mensagem a ser usada como mensagem inicial na exportação.
Para obter mais informações sobre `MessageStreamDefinition`, consulte a referência do SDK para a sua linguagem de destino:
+ [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)no Java SDK
+ [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)no SDK do Node.js
+ [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)no SDK do Python
**nota**
O `StreamManagerClient` também fornece um destino alvo que você pode usar para exportar fluxos para um servidor HTTP. Este destino deve ser usado apenas para fins de teste. Ele não é estável e nem compatível para uso em ambientes de produção.
Depois que um fluxo é criado, os componentes do Greengrass podem [anexar mensagens](#streammanagerclient-append-message) ao fluxo para enviar dados para exportação e [ler mensagens](#streammanagerclient-read-messages) do fluxo para processamento local. O número de fluxos criados depende dos seus recursos de hardware e caso de negócios. Uma estratégia é criar um stream para cada canal de destino no stream AWS IoT Analytics de dados do Kinesis, embora você possa definir vários destinos para um stream. Um fluxo tem longa duração.
### Requisitos
Essa operação tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Exemplos
O snippet a seguir cria um fluxo chamado `StreamName`. Ele define as propriedades de fluxo em `MessageStreamDefinition` e nos tipos de dados subordinados.
------
#### [ 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 Nuvem AWS.
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.
```
[Referência do SDK do Python: create\$1message\$1stream \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.create_message_stream)
------
#### [ Java ]
```
try (final StreamManagerClient client = StreamManagerClientFactory.standard().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 Nuvem AWS.
new ExportDefinition()
.withKinesis(null)
.withIotAnalytics(null)
.withIotSitewise(null)
.withS3(null)
)
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Referência do SDK Java: \$1 [createMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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 Nuvem AWS.
new ExportDefinition()
.withKinesis(null)
.withIotAnalytics(null)
.withIotSiteWise(null)
.withS3(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.
});
```
Referência do SDK do Node.js: \$1 [createMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#createMessageStream)
------
Para obter mais informações sobre como configurar destinos de exportação, consulte [Configurações de exportação para destinos compatíveis Nuvem AWS](stream-export-configurations.md).
## Anexar mensagem
Para enviar dados ao Gerenciador de fluxos para exportação, os componentes do Greengrass anexam os dados ao fluxo de destino. O destino da exportação determina o tipo de dados a ser passado para esse método.
### Requisitos
Essa operação tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Exemplos
#### AWS IoT Analytics ou destinos de exportação do Kinesis Data Streams
O snippet a seguir anexa uma mensagem ao fluxo chamado `StreamName`. Para AWS IoT Analytics nossos destinos do Kinesis Data Streams, seus componentes do Greengrass acrescentam um blob de dados.
Esse snippet tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.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.
```
Referência do SDK em Python: [append\$1message](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message)
------
#### [ Java ]
```
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) {
long sequenceNumber = client.appendMessage("StreamName", "Arbitrary byte array".getBytes());
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Referência do SDK em Java: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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.
});
```
Referência do SDK em Node.js: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage)
------
#### AWS IoT SiteWise destinos de exportação
O snippet a seguir anexa uma mensagem ao fluxo chamado `StreamName`. Para AWS IoT SiteWise destinos, seus componentes do Greengrass acrescentam um objeto serializado. `PutAssetPropertyValueEntry` Para obter mais informações, consulte [Exportando para AWS IoT SiteWise](stream-export-configurations.md#export-streams-to-sitewise).
**nota**
Quando você envia dados para AWS IoT SiteWise, seus dados devem atender aos requisitos da `BatchPutAssetPropertyValue` ação. Para obter mais informações, consulte [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) na *Referência de APIs do AWS IoT SiteWise *.
Esse snippet tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
------
#### [ Python ]
```
client = StreamManagerClient()
try:
# SiteWise requires unique timestamps in all messages and also needs timestamps not earlier
# than 10 minutes in the past. 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", Util.validate_and_serialize_to_json_bytes(putAssetPropertyValueEntry))
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
[Referência do SDK do Python: append\$1message \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.PutAssetPropertyValueEntry)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#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 and also needs timestamps not earlier
// than 10 minutes in the past. 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.
}
```
Referência do SDK Java: [appendMessage \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/sitewise/PutAssetPropertyValueEntry.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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.
});
```
Referência do SDK do Node.js: [appendMessage \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.PutAssetPropertyValueEntry.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage)
------
#### Destinos de exportação do Amazon S3
O snippet a seguir anexa uma tarefa de exportação ao fluxo chamada `StreamName`. Para os destinos do Amazon S3, os componentes do Greengrass anexam um objeto serializado `S3ExportTaskDefinition` que contém informações sobre o arquivo de entrada de origem e o objeto do Amazon S3 de destino. Se o objeto especificado não existir, o gerenciador de fluxo criará o objeto para você. Para obter mais informações, consulte [Exportar para o Amazon S3](stream-export-configurations.md#export-streams-to-s3).
Esse snippet tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.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", 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.
```
[https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message)
------
#### [ 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.
}
```
[Referência do SDK Java: [appendMessage \$1](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) S3 ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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.
});
```
[Referência do SDK do Node.js: [appendMessage \$1](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) S3 ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskDefinition.html)
------
## Ler Mensagens
Ler mensagens de um fluxo.
### Requisitos
Essa operação tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Exemplos
O snippet a seguir lê mensagens do fluxo chamado `StreamName`. O método de leitura usa um objeto `ReadMessagesOptions` opcional que especifica o número de sequência a partir do qual começar a ler, os números mínimo e máximo a ler e um tempo limite para ler mensagens.
------
#### [ 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.
```
[Referência do SDK do Python: read\$1messages \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.ReadMessagesOptions)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.read_messages)
------
#### [ Java ]
```
try (final StreamManagerClient client = StreamManagerClientFactory.standard().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.
}
```
Referência do SDK Java: [readMessages \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/ReadMessagesOptions.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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.
});
```
Referência do SDK do Node.js: [readMessages \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.ReadMessagesOptions.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages)
------
## Listar fluxos
Obtenha a lista de fluxos no gerenciador de fluxos.
### Requisitos
Essa operação tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Exemplos
O snippet a seguir obtém uma lista dos fluxos (por nome) no gerenciador de fluxo.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
stream_names = client.list_streams()
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
Referência do SDK em Python: [list\$1streams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.list_streams)
------
#### [ Java ]
```
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) {
List streamNames = client.listStreams();
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Referência do SDK em Java: [ListStreams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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.
});
```
Referência do SDK em Node.js: [listStreams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#listStreams)
------
## Descrever stream de mensagens
Obtenha metadados sobre um fluxo, incluindo a definição, o tamanho e o status de exportação do fluxo.
### Requisitos
Essa operação tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Exemplos
O snippet a seguir obtém metadados sobre o fluxo chamado `StreamName`, incluindo a definição, o tamanho e o status do exportador do fluxo.
------
#### [ 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.
```
Referência do SDK em Python: [describe\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.describe_message_stream)
------
#### [ Java ]
```
try (final StreamManagerClient client = StreamManagerClientFactory.standard().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.
}
```
Referência do Java SDK: [describeMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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.
});
```
Referência do SDK do Node.js: [describeMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#describeMessageStream)
------
## Atualize o fluxo de mensagens
Atualize as propriedades de um fluxo existente. Talvez você queira atualizar um fluxo se seus requisitos mudarem após a criação do fluxo. Por exemplo:
+ Adicione uma nova [configuração de exportação](stream-export-configurations.md) para um Nuvem AWS destino.
+ Aumente o tamanho máximo de um fluxo para alterar a forma como os dados são exportados ou retidos. Por exemplo, o tamanho do fluxo em combinação com sua estratégia em configurações completas pode resultar na exclusão ou rejeição dos dados antes que o gerenciador de fluxo possa processá-los.
+ Pause e retome as exportações; por exemplo, se as tarefas de exportação forem demoradas e você quiser racionar seus dados de upload.
Os componentes do Greengrass seguem este processo de alto nível para atualizar um fluxo:
1. [Obter a descrição do fluxo.](#streammanagerclient-describe-message-stream)
1. Atualizar as propriedades de destino nos objetos correspondentes `MessageStreamDefinition` e subordinados.
1. Passar o atualizado `MessageStreamDefinition`. Certifique-se de incluir as definições completas do objeto para o fluxo atualizado. As propriedades indefinidas revertem para os valores padrão.
Você pode especificar o número de sequência da mensagem a ser usada como mensagem inicial na exportação.
### Requisitos
Essa operação tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Exemplos
O snippet a seguir atualiza o fluxo chamado `StreamName`. Ele atualiza várias propriedades de um fluxo que é exportado para o 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.
```
Referência do SDK do Python: \$1 [updateMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#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 Nuvem AWS.
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.
}
```
Referência do SDK Java: [update\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#updateMessageStream-java.lang.String-) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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 Nuvem AWS.
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.
});
```
Referência do SDK do Node.js: \$1 [updateMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#updateMessageStream)
------
### Restrições para a atualização de fluxos
As restrições a seguir se aplicam ao atualizar fluxos. A menos que indicado na lista a seguir, as atualizações entrarão em vigor imediatamente.
+ Não é possível atualizar a persistência de um fluxo. Para alterar esse comportamento, [exclua o fluxo](#streammanagerclient-delete-message-stream) e [crie um fluxo](#streammanagerclient-create-message-stream) que defina a nova política de persistência.
+ Você só pode atualizar o tamanho máximo de um fluxo sob as seguintes condições:
+ O tamanho máximo deve ser maior que o tamanho atual do fluxo. Para encontrar essas informações, [descreva o fluxo](#streammanagerclient-describe-message-stream) e, em seguida, verifique o status de armazenamento do objeto `MessageStreamInfo` retornado.
+ O tamanho máximo deve ser maior ou igual ao tamanho do segmento do fluxo.
+ Você pode atualizar o tamanho do segmento do fluxo para um valor menor que o tamanho máximo do fluxo. A configuração atualizada se aplica aos novos segmentos.
+ As atualizações da propriedade tempo de vida (TTL) se aplicam às novas operações de anexação. Se você diminuir esse valor, o gerenciador de fluxo também poderá excluir segmentos existentes que excedam o TTL.
+ As atualizações da estratégia em toda a propriedade se aplicam às novas operações de anexação. Se você definir a estratégia para substituir os dados mais antigos, o gerenciador de fluxo também poderá substituir os segmentos existentes com base na nova configuração.
+ As atualizações na propriedade “descartar após gravação” se aplicam às novas mensagens.
+ As atualizações nas configurações de exportação se aplicam às novas exportações. A solicitação de atualização deve incluir todas as configurações de exportação às quais você deseja oferecer suporte. Caso contrário, o gerenciador de fluxo as excluirá.
+ Ao atualizar uma configuração de exportação, especifique o identificador da configuração de exportação de destino.
+ Para adicionar uma configuração de exportação, especifique um identificador exclusivo para a nova configuração de exportação.
+ Para excluir uma configuração de exportação, omita a configuração de exportação.
+ Para [atualizar](#streammanagerclient-update-message-stream) o número da sequência inicial de uma configuração de exportação em um fluxo, você deve especificar um valor menor que o número de sequência mais recente. Para encontrar essas informações, [descreva o fluxo](#streammanagerclient-describe-message-stream) e, em seguida, verifique o status de armazenamento do objeto `MessageStreamInfo` retornado.
## Excluir stream de mensagens
Exclui um fluxo. Quando você exclui um fluxo, todos os dados armazenados para o fluxo são excluídos do disco.
### Requisitos
Essa operação tem os seguintes requisitos:
+ Versão mínima do Stream Manager SDK: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Exemplos
O snippet a seguir exclui o fluxo chamado `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.
```
Referência do SDK do Python: [deleteMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.delete_message_stream)
------
#### [ Java ]
```
try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) {
client.deleteMessageStream("StreamName");
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Referência do SDK em Java: [delete\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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.
});
```
Referência do SDK do Node.js: [deleteMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#deleteMessageStream)
------
## Consulte também
+ [Gerenciar fluxos de dados no nos dispositivos principais do Greengrass](manage-data-streams.md)
+ [Configurar o gerenciador de fluxo do AWS IoT Greengrass](configure-stream-manager.md)
+ [Configurações de exportação para destinos compatíveis Nuvem AWS](stream-export-configurations.md)
+ `StreamManagerClient` na referência do Stream Manager SDK:
+ [Python](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html)
+ [Java](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html)
+ [Node.js](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html)
# Configurações de exportação para destinos compatíveis Nuvem AWS
Os componentes do Greengrass definidos pelo usuário usam o `StreamManagerClient` no Stream Manager SDK para interagir com o Gerenciador de fluxos. Quando um componente [cria um fluxo](work-with-streams.md#streammanagerclient-create-message-stream) ou [atualiza um fluxo](work-with-streams.md#streammanagerclient-create-message-stream), ele passa um objeto `MessageStreamDefinition` que representa as propriedades do fluxo, incluindo a definição de exportação. O objeto `ExportDefinition` contém as configurações de exportação definidas para o fluxo. O gerenciador de fluxo usa essas configurações de exportação para determinar onde e como exportar o fluxo.
![\[Diagrama do modelo de objeto do tipo de ExportDefinition propriedade.\]](http://docs.aws.amazon.com/pt_br/greengrass/v2/developerguide/images/stream-manager-exportconfigs.png)
É possível definir zero ou mais configurações de exportação em um fluxo, incluindo várias configurações de exportação para um único tipo de destino. Por exemplo, você pode exportar um fluxo para dois canais do AWS IoT Analytics e um fluxo de dados do Kinesis.
Para tentativas de exportação malsucedidas, o stream manager tenta continuamente exportar dados para o Nuvem AWS em intervalos de até cinco minutos. Não há um limite máximo para o número de novas tentativas.
**nota**
O `StreamManagerClient` também fornece um destino alvo que você pode usar para exportar fluxos para um servidor HTTP. Este destino deve ser usado apenas para fins de teste. Ele não é estável e nem compatível para uso em ambientes de produção.
**Topics**
+ [AWS IoT Analytics canais](#export-to-iot-analytics)
+ [Amazon Kinesis Data Streams](#export-to-kinesis)
+ [AWS IoT SiteWise propriedades do ativo](#export-to-iot-sitewise)
+ [Objetos do Amazon S3](#export-to-s3)
Você é responsável pela manutenção desses Nuvem AWS recursos.
## AWS IoT Analytics canais
O gerenciador de fluxo suporta exportações automáticas para AWS IoT Analytics. AWS IoT Analytics permite realizar análises avançadas em seus dados para ajudar a tomar decisões comerciais e melhorar os modelos de aprendizado de máquina. Para obter mais informações, consulte [O que é AWS IoT Analytics?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) no *Guia do AWS IoT Analytics usuário*.
No Stream Manager SDK, os componentes do Greengrass usam `IoTAnalyticsConfig` para definir a configuração de exportação para esse tipo de destino. Para mais informações, consulte a referência do SDK para seu idioma de destino:
+ [Io TAnalytics Config no SDK](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTAnalyticsConfig) do Python
+ [Io TAnalytics Config no Java](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTAnalyticsConfig.html) SDK
+ [Io TAnalytics Config no SDK](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTAnalyticsConfig.html) do Node.js
### Requisitos
Esse destino de exportação tem os seguintes requisitos:
+ Os canais de destino AWS IoT Analytics devem estar no mesmo dispositivo principal Conta da AWS e Região da AWS no dispositivo principal do Greengrass.
+ O [Autorize os dispositivos principais a interagir com os serviços AWS](device-service-role.md) deve conceder a permissão `iotanalytics:BatchPutMessage` para os canais de destino. Por exemplo:
------
#### [ 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"
]
}
]
}
```
------
É possível conceder acesso granular ou condicional aos recursos, por exemplo, usando um esquema de nomeação `*` curinga. Para obter mais informações, consulte [Adicionando e removendo políticas do IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) no *Guia do usuário do IAM*.
### Exportando para AWS IoT Analytics
Para criar um fluxo que exporta para AWS IoT Analytics, seus componentes do Greengrass [criam um fluxo](work-with-streams.md#streammanagerclient-create-message-stream) com uma definição de exportação que inclui um ou mais `IoTAnalyticsConfig` objetos. Esse objeto define as configurações de exportação, como canal de destino, tamanho do lote, intervalo do lote e prioridade.
Quando os componentes do Greengrass recebem dados de dispositivos, eles [anexam mensagens](work-with-streams.md#streammanagerclient-append-message) que contêm um blob de dados ao fluxo de destino.
Em seguida, o gerenciador de fluxo exporta os dados com base nas configurações de lote e na prioridade definidas nas configurações de exportação do fluxo.
## Amazon Kinesis Data Streams
O gerenciador de fluxos é compatível com exportações automáticas para o Amazon Kinesis Data Streams. O Kinesis Data Streams é comumente usado para agregar dados de alto volume e carregá-los em um data warehouse ou cluster. MapReduce Para obter mais informações, consulte [O que é o Amazon Kinesis Data Streams?](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) no *Guia do desenvolvedor do Amazon Kinesis*.
No Stream Manager SDK, os componentes do Greengrass usam `KinesisConfig` para definir a configuração de exportação para esse tipo de destino. Para mais informações, consulte a referência do SDK para seu idioma de destino:
+ [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.KinesisConfig)no SDK do Python
+ [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/KinesisConfig.html)no Java SDK
+ [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.KinesisConfig.html)no SDK do Node.js
### Requisitos
Esse destino de exportação tem os seguintes requisitos:
+ Os streams de destino no Kinesis Data Streams devem estar no mesmo dispositivo principal do Conta da AWS Região da AWS Greengrass.
+ (Recomendado) O Gerenciador de fluxos v2.2.1 melhora a performance da exportação de fluxos para destinos do Kinesis Data Streams. Para usar as melhorias nesta versão mais recente, atualize seu [componente do gerenciador de fluxos](stream-manager-component.md) para a versão 2.2.1 e use a política `kinesis:ListShards` no [perfil de troca de tokens do Greengrass](device-service-role.md).
+ O [Autorize os dispositivos principais a interagir com os serviços AWS](device-service-role.md) deve conceder a permissão `kinesis:PutRecords` para os fluxos de dados de destino. Por exemplo:
------
#### [ 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"
]
}
]
}
```
------
É possível conceder acesso granular ou condicional aos recursos, por exemplo, usando um esquema de nomeação `*` curinga. Para obter mais informações, consulte [Adicionando e removendo políticas do IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) no *Guia do usuário do IAM*.
### Exportação do Kinesis Data Streams
Para criar um fluxo que exporte para o Kinesis Data Streams, os componentes do Greengrass [criam um fluxo](work-with-streams.md#streammanagerclient-create-message-stream) com uma definição de exportação que inclui um ou mais objetos `KinesisConfig`. Esse objeto define as configurações de exportação, como fluxo de dados, tamanho do lote, intervalo do lote e prioridade.
Quando os componentes do Greengrass recebem dados de dispositivos, eles [anexam mensagens](work-with-streams.md#streammanagerclient-append-message) que contêm um blob de dados ao fluxo de destino. Em seguida, o gerenciador de fluxo exporta os dados com base nas configurações de lote e na prioridade definidas nas configurações de exportação do fluxo.
O gerenciador de fluxo gera uma UUID exclusiva e aleatória como chave de partição para cada registro carregado no Amazon Kinesis.
## AWS IoT SiteWise propriedades do ativo
O gerenciador de fluxo suporta exportações automáticas para AWS IoT SiteWise. AWS IoT SiteWise permite coletar, organizar e analisar dados de equipamentos industriais em grande escala. Para obter mais informações, consulte [O que é AWS IoT SiteWise?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) no *Guia do AWS IoT SiteWise usuário*.
No Stream Manager SDK, os componentes do Greengrass usam `IoTSiteWiseConfig` para definir a configuração de exportação para esse tipo de destino. Para mais informações, consulte a referência do SDK para seu idioma de destino:
+ [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTSiteWiseConfig) no SDK do Python
+ [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTSiteWiseConfig.html) no Java SDK
+ [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTSiteWiseConfig.html) no SDK do Node.js
**nota**
AWS também fornece AWS IoT SiteWise componentes, que oferecem uma solução pré-construída que você pode usar para transmitir dados de fontes OPC-UA. Para obter mais informações, consulte [Coletor IoT SiteWise OPC UA](iotsitewise-opcua-collector-component.md).
### Requisitos
Esse destino de exportação tem os seguintes requisitos:
+ As propriedades do ativo de destino AWS IoT SiteWise devem estar no mesmo Conta da AWS Região da AWS dispositivo central do Greengrass.
**nota**
Para ver a lista de Região da AWS s que AWS IoT SiteWise oferecem suporte, consulte [AWS IoT SiteWise endpoints e cotas](https://docs.aws.amazon.com/general/latest/gr/iot-sitewise.html#iot-sitewise_region) na Referência *AWS geral*.
+ O [Autorize os dispositivos principais a interagir com os serviços AWS](device-service-role.md) deve conceder a permissão `iotsitewise:BatchPutAssetPropertyValue` para as propriedades do ativo do destino. O exemplo de política a seguir usa a chave de condição `iotsitewise:assetHierarchyPath` para conceder acesso a um ativo raiz de destino e seus ativos secundários. Você pode remover o `Condition` da política para permitir o acesso a todos os seus AWS IoT SiteWise ativos ou especificar ARNs ativos individuais.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "iotsitewise:BatchPutAssetPropertyValue",
"Resource": "*",
"Condition": {
"StringLike": {
"iotsitewise:assetHierarchyPath": [
"/root node asset ID",
"/root node asset ID/*"
]
}
}
}
]
}
```
------
É possível conceder acesso granular ou condicional aos recursos, por exemplo, usando um esquema de nomeação `*` curinga. Para obter mais informações, consulte [Adicionando e removendo políticas do IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) no *Guia do usuário do IAM*.
Para obter informações de segurança importantes, consulte a [ BatchPutAssetPropertyValue autorização](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) no *Guia AWS IoT SiteWise do usuário*.
### Exportando para AWS IoT SiteWise
Para criar um fluxo que exporta para AWS IoT SiteWise, seus componentes do Greengrass [criam um fluxo](work-with-streams.md#streammanagerclient-create-message-stream) com uma definição de exportação que inclui um ou mais `IoTSiteWiseConfig` objetos. Esse objeto define as configurações de exportação, como tamanho do lote, intervalo do lote e prioridade.
Quando os componentes do Greengrass recebem dados de propriedades de ativos dos dispositivos, eles anexam mensagens que contêm um blob de dados ao fluxo de destino. As mensagens são objetos `PutAssetPropertyValueEntry` serializados em JSON que contêm valores de propriedade para uma ou mais propriedades de ativos. Para obter mais informações, consulte [Anexar mensagem](work-with-streams.md#streammanagerclient-append-message-sitewise) para destinos de exportação do AWS IoT SiteWise .
**nota**
Quando você envia dados para AWS IoT SiteWise, seus dados devem atender aos requisitos da `BatchPutAssetPropertyValue` ação. Para obter mais informações, consulte [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) na *Referência de APIs do AWS IoT SiteWise *.
Em seguida, o gerenciador de fluxo exporta os dados com base nas configurações de lote e na prioridade definidas nas configurações de exportação do fluxo.
É possível ajustar as configurações do Gerenciador de fluxos e a lógica do componente do Greengrass para criar sua estratégia de exportação. Por exemplo:
+ Para exportações quase em tempo real, defina configurações baixas de tamanho de lote e intervalo e anexe os dados ao fluxo quando forem recebidos.
+ Para otimizar o agrupamento em lotes, mitigar as restrições de largura de banda ou minimizar os custos, seus componentes do Greengrass podem agrupar os pontos de dados timestamp-quality-value (TQV) recebidos para uma única propriedade do ativo antes de anexar os dados ao stream. Uma estratégia é agrupar entradas para até 10 (dez) combinações diferentes de propriedade e ativo, ou aliases de propriedade, em uma mensagem, em vez de enviar mais de uma entrada para a mesma propriedade. Isso ajuda o gerenciador de fluxo a permanecer dentro das [cotas do AWS IoT SiteWise](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/quotas.html).
## Objetos do Amazon S3
O gerenciador de fluxo é compatível com exportações automáticas para o Amazon S3. É possível utilizar o Amazon S3 para armazenar e recuperar grandes volumes de dados. Para obter mais informações, consulte [O que é o Amazon S3?](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) no *Guia do desenvolvedor do Amazon Simple Storage Service*.
No Stream Manager SDK, os componentes do Greengrass usam `S3ExportTaskExecutorConfig` para definir a configuração de exportação para esse tipo de destino. Para mais informações, consulte a referência do SDK para seu idioma de destino:
+ [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.S3ExportTaskExecutorConfig) no SDK do Python
+ [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/S3ExportTaskExecutorConfig.html) no Java SDK
+ [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskExecutorConfig.html) no SDK do Node.js
### Requisitos
Esse destino de exportação tem os seguintes requisitos:
+ Os buckets do Amazon S3 de destino devem estar no mesmo dispositivo principal Conta da AWS do Greengrass.
+ Se uma função do Lambda executada no modo de **contêiner do Greengrass** grava arquivos de entrada no diretório de arquivos de entrada, é necessário montar o diretório como um volume no contêiner com permissões de gravação. Esse procedimento garante que os arquivos sejam gravados no sistema de arquivos raiz e fiquem visíveis para o componente Gerenciador de fluxos, executado fora do contêiner.
+ Se um componente de contêiner do Docker grava arquivos de entrada no diretório de arquivos de entrada, é necessário montar o diretório como um volume no contêiner com permissões de gravação. Esse procedimento garante que os arquivos sejam gravados no sistema de arquivos raiz e fiquem visíveis para o componente Gerenciador de fluxos, executado fora do contêiner.
+ O [Autorize os dispositivos principais a interagir com os serviços AWS](device-service-role.md) deve conceder as permissões a seguir para os buckets de destino. Por exemplo:
------
#### [ 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/*"
]
}
]
}
```
------
É possível conceder acesso granular ou condicional aos recursos, por exemplo, usando um esquema de nomeação `*` curinga. Para obter mais informações, consulte [Adicionando e removendo políticas do IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) no *Guia do usuário do IAM*.
### Exportar para o Amazon S3
Para criar um fluxo que exporte para o Amazon S3, os componentes do Greengrass usam o objeto `S3ExportTaskExecutorConfig` para configurar a política de exportação. A política define as configurações de exportação, como o limite e a prioridade de upload em várias partes. Para exportações do Amazon S3, o gerenciador de fluxo carrega dados que ele lê de arquivos locais no dispositivo principal. Para iniciar um upload, os componentes do Greengrass anexam uma tarefa de exportação ao fluxo de destino. A tarefa de exportação contém informações sobre o arquivo de entrada e o objeto de destino do Amazon S3. O Gerenciador de fluxos executa as tarefas na sequência em que elas são anexadas ao fluxo.
**nota**
O bucket de destino já deve existir no seu Conta da AWS. Se um objeto para a chave especificada não existir, o gerenciador de fluxo criará o objeto para você.
O gerenciador de fluxo usa a propriedade de limite de upload de várias partes, a configuração do [tamanho mínimo das partes](configure-stream-manager.md#stream-manager-minimum-part-size) e o tamanho do arquivo de entrada para determinar como fazer upload dos dados. O limite de upload de várias partes deve ser maior que o tamanho mínimo das partes. Se você quiser fazer upload de dados em paralelo, pode criar vários fluxos.
As chaves que especificam seus objetos de destino do Amazon S3 podem incluir DateTimeFormatter cadeias de caracteres [Java](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html) válidas em espaços reservados. `!{timestamp:value}` É possível usar esses espaços reservados de data e hora para particionar dados no Amazon S3 com base na hora em que os dados do arquivo de entrada foram carregados. Por exemplo, o nome da chave a seguir é resolvido para um valor como `my-key/2020/12/31/data.txt`.
```
my-key/!{timestamp:YYYY}/!{timestamp:MM}/!{timestamp:dd}/data.txt
```
**nota**
Se você quiser monitorar o status de exportação de um fluxo, primeiro crie um fluxo de status e, em seguida, configure o fluxo de exportação para usá-lo. Para obter mais informações, consulte [Monitorar tarefas de exportação](#monitor-export-status-s3).
#### Gerenciar dados de entrada
É possível criar códigos que as aplicações de IoT usam para gerenciar o ciclo de vida dos dados de entrada. O exemplo de fluxo de trabalho a seguir mostra como você pode usar os componentes do Greengrass para gerenciar esses dados.
1. Um processo local recebe dados de dispositivos ou periféricos e, em seguida, grava os dados em arquivos em um diretório no dispositivo principal. Esses são os arquivos de entrada para o gerenciador de fluxo.
1. O componente do Greengrass verifica o diretório e [anexa uma tarefa de exportação](work-with-streams.md#streammanagerclient-append-message-export-task) ao fluxo de destino quando um novo arquivo é criado. A tarefa é um objeto `S3ExportTaskDefinition` serializado em JSON que especifica a URL do arquivo de entrada, o bucket e a chave do Amazon S3 de destino, além dos metadados opcionais do usuário.
1. O gerenciador de fluxo lê o arquivo de entrada e exporta os dados para o Amazon S3 na ordem das tarefas anexadas. O bucket de destino já deve existir no seu Conta da AWS. Se um objeto para a chave especificada não existir, o gerenciador de fluxo criará o objeto para você.
1. O componente do Greengrass [lê as mensagens](work-with-streams.md#streammanagerclient-read-messages) de um fluxo de status para monitorar o status da exportação. Depois que as tarefas de exportação forem concluídas, o componente do Greengrass poderá excluir os arquivos de entrada correspondentes. Para obter mais informações, consulte [Monitorar tarefas de exportação](#monitor-export-status-s3).
### Monitorar tarefas de exportação
É possível criar códigos que as aplicações de IoT usam para monitorar o status das suas exportações do Amazon S3. Os componentes do Greengrass devem criar um fluxo de status e configurar o fluxo de exportação para gravar as atualizações de status no fluxo de status. Um único fluxo de status pode receber atualizações de status de vários fluxos que são exportados para o Amazon S3.
Primeiro, [crie um fluxo](work-with-streams.md#streammanagerclient-create-message-stream) para usar como fluxo de status. É possível configurar as políticas de tamanho e retenção do fluxo para controlar a vida útil das mensagens de status. Por exemplo:
+ Defina `Persistence` como `Memory` se você não quiser armazenar as mensagens de status.
+ Defina `StrategyOnFull` como `OverwriteOldestData` para que as novas mensagens de status não sejam perdidas.
Em seguida, crie ou atualize o fluxo de exportação para usar o fluxo de status. Especificamente, defina a propriedade de configuração de status da configuração de exportação `S3ExportTaskExecutorConfig` do fluxo. Essa configuração instrui o Gerenciador de fluxos a gravar as mensagens de status sobre as tarefas de exportação no fluxo de status. No objeto `StatusConfig`, especifique o nome do fluxo de status e o nível de detalhe. Os valores com suporte a seguir variam do menos detalhado (`ERROR`) ao mais detalhado (`TRACE`). O padrão é `INFO`.
+ `ERROR`
+ `WARN`
+ `INFO`
+ `DEBUG`
+ `TRACE`
O exemplo de fluxo de trabalho a seguir mostra como os componentes do Greengrass podem usar um fluxo de status para monitorar o status da exportação.
1. Conforme descrito no fluxo de trabalho anterior, um componente do [anexa uma tarefa de exportação](work-with-streams.md#streammanagerclient-append-message-export-task) a um fluxo configurado para gravar mensagens de status sobre as tarefas de exportação em um fluxo de status. A operação de append retorna um número de sequência que representa a ID da tarefa.
1. U componente do Greengrass [lê as mensagens](work-with-streams.md#streammanagerclient-read-messages) sequencialmente do fluxo de status e as filtra com base no nome do fluxo e no ID da tarefa, ou com base em uma propriedade da tarefa de exportação do contexto da mensagem. Por exemplo, o componente do Greengrass pode filtrar por URL do arquivo de entrada da tarefa de exportação, que é representada pelo objeto `S3ExportTaskDefinition` no contexto da mensagem.
Os códigos de status a seguir indicam que uma tarefa de exportação atingiu um estado concluído:
+ `Success`. O upload foi concluído com êxito.
+ `Failure`. O gerenciador de fluxo encontrou um erro, por exemplo, o bucket especificado não existe. Depois de resolver o problema, você pode reanexar a tarefa de exportação ao fluxo.
+ `Canceled`. A tarefa foi interrompida porque a definição de fluxo ou exportação foi excluída ou o período time-to-live (TTL) da tarefa expirou.
**nota**
A tarefa também pode ter um status de `InProgress` ou `Warning`. O gerenciador de fluxo emite avisos quando um evento retorna um erro que não afeta a execução da tarefa. Por exemplo, uma falha na limpeza de um upload parcial retorna um aviso.
1. Depois que as tarefas de exportação forem concluídas, o componente do Greengrass poderá excluir os arquivos de entrada correspondentes.
O exemplo a seguir mostra como um componente do Greengrass pode ler e processar as mensagens de status.
------
#### [ Python ]
```
import time
from stream_manager import (
ReadMessagesOptions,
Status,
StatusConfig,
StatusLevel,
StatusMessage,
StreamManagerClient,
)
from 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.
```
[Referência do SDK do Python: read\$1messages \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.StatusMessage)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.read_messages)
------
#### [ Java ]
```
import com.amazonaws.greengrass.streammanager.client.StreamManagerClient;
import com.amazonaws.greengrass.streammanager.client.StreamManagerClientFactory;
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 = StreamManagerClientFactory.standard().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.
}
```
Referência do SDK Java: [readMessages \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/StatusMessage.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-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-stream-manager-sdk'*);
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.
});
```
Referência do SDK do Node.js: [readMessages \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StatusMessage.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages)
------