Pemberitahuan akhir dukungan: Pada 7 Oktober 2026, AWS akan menghentikan dukungan untuk. AWS IoT Greengrass Version 1 Setelah 7 Oktober 2026, Anda tidak akan lagi dapat mengakses sumber daya. AWS IoT Greengrass V1 Untuk informasi lebih lanjut, silakan kunjungi [Migrasi dari AWS IoT Greengrass Version 1](https://docs.aws.amazon.com/greengrass/v2/developerguide/migrate-from-v1.html).
Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Gunakan StreamManagerClient untuk bekerja dengan aliran
Fungsi Lambda yang ditentukan pengguna yang berjalan pada AWS IoT Greengrass inti dapat menggunakan `StreamManagerClient` objek di [AWS IoT Greengrass Core SDK](lambda-functions.md#lambda-sdks) untuk membuat aliran di [pengelola aliran dan kemudian berinteraksi dengan aliran](stream-manager.md). Saat fungsi Lambda membuat aliran, fungsi ini akan menentukan AWS Cloud tujuan, prioritas, dan kebijakan ekspor dan penyimpanan data lainnya untuk aliran. Untuk mengirim data ke pengelola pengaliran, fungsi Lambda menambahkan data ke pengaliran. Jika tujuan ekspor ditentukan untuk pengaliran, pengelola pengaliran mengekspor pengaliran secara otomatis.
**catatan**
Biasanya, klien dari pengelola pengaliran adalah fungsi Lambda yang ditetapkan pengguna. Jika kasus bisnis Anda memerlukannya, Anda juga dapat mengizinkan proses non-Lambda berjalan pada core Greengrass (sebagai contoh, wadah Docker) untuk berinteraksi dengan pengelola pengaliran. Untuk informasi selengkapnya, lihat [Autentikasi klien](stream-manager.md#stream-manager-security-client-authentication).
Cuplikan dalam topik ini menunjukkan cara klien memanggil metode `StreamManagerClient` untuk bekerja dengan aliran. Untuk rincian implementasi tentang metode dan argumen mereka, gunakan tautan ke referensi SDK yang tercantum setelah setiap cuplikan. Untuk tutorial yang mencakup fungsi Python Lambda lengkap, lihat [Ekspor aliran data ke AWS Cloud (konsol)](stream-manager-console.md) atau [Ekspor aliran data ke AWS Cloud (CLI)](stream-manager-cli.md).
Fungsi Lambda Anda harus berjalan `StreamManagerClient` di luar dari fungsi handler. Jika berjalan di dalam handler, fungsi membuat `client` dan koneksi ke pengelola pengaliran setiap kali dipanggil.
**catatan**
Jika Anda membuat contoh `StreamManagerClient` dalam handler, Anda harus secara tegas memanggil metode `close()` ketika `client` menyelesaikan pekerjaannya. Jika tidak, `client` akan membuat sambungan terbuka dan utas lain yang berjalan sampai skrip keluar.
`StreamManagerClient` mendukung operasi berikut:
+ [Membuat pengaliran pesan](#streammanagerclient-create-message-stream)
+ [Tambahkan pesan](#streammanagerclient-append-message)
+ [Baca pesan](#streammanagerclient-read-messages)
+ [Daftar aliran](#streammanagerclient-list-streams)
+ [Jelaskan aliran pesan](#streammanagerclient-describe-message-stream)
+ [Perbarui aliran pesan](#streammanagerclient-update-message-stream)
+ [Hapus aliran pesan](#streammanagerclient-delete-message-stream)
## Membuat pengaliran pesan
Untuk membuat pengaliran, fungsi Lambda yang ditetapkan pengguna memanggil metode pembuatan dan melewati di dalam `MessageStreamDefinition` objek. Objek ini menentukan nama unik untuk pengaliran dan menentukan cara pengelola pengaliran harus menangani data baru ketika ukuran pengaliran maksimum tercapai. Anda dapat menggunakan `MessageStreamDefinition` dan jenis datanya (seperti `ExportDefinition`, `StrategyOnFull`, dan `Persistence`) untuk menentukan properti pengaliran lainnya. Ini termasuk:
+ Target AWS IoT Analytics, Kinesis Data AWS IoT SiteWise Streams, dan tujuan Amazon S3 untuk ekspor otomatis. Untuk informasi selengkapnya, lihat [Ekspor konfigurasi untuk tujuan yang didukung AWS Cloud](stream-export-configurations.md).
+ Prioritas ekspor. Stream manager mengekspor aliran prioritas yang lebih tinggi sebelum aliran prioritas lebih rendah.
+ Ukuran batch maksimum dan interval batch untuk AWS IoT Analytics, Kinesis Data Streams AWS IoT SiteWise , dan tujuan. Stream manager mengekspor pesan ketika salah satu kondisi terpenuhi.
+ Time-to-live (TTL). Jumlah waktu untuk menjamin bahwa data aliran tersedia untuk diproses. Anda harus memastikan bahwa data dapat dikonsumsi dalam periode waktu ini. Ini bukan kebijakan penghapusan. Data mungkin tidak segera dihapus setelah periode TTL.
+ Ketekunan aliran. Pilih untuk menyimpan aliran ke sistem file untuk mempertahankan data di seluruh restart inti atau menyimpan aliran dalam memori.
+ Memulai nomor urut. Tentukan nomor urutan pesan yang akan digunakan sebagai pesan awal dalam ekspor.
Untuk informasi lebih lanjut tentang `MessageStreamDefinition`, lihat referensi SDK untuk bahasa target Anda:
+ [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)di Java SDK
+ [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)di Node.js SDK
+ [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.MessageStreamDefinition)di Python SDK
**catatan**
`StreamManagerClient` juga menyediakan target tujuan yang dapat Anda gunakan untuk mengekspor aliran ke server HTTP. Target ini ditujukan untuk tujuan pengujian saja. Hal ini tidak stabil atau didukung untuk digunakan di lingkungan produksi.
Setelah pengaliran dibuat, fungsi Lambda Anda dapat [tambahkan pesan](#streammanagerclient-append-message) ke pengaliran untuk mengirim data untuk ekspor dan [baca pesan](#streammanagerclient-append-message) dari pengaliran untuk pemrosesan lokal. Jumlah pengaliran yang Anda buat tergantung pada kemampuan perangkat keras dan kasus bisnis Anda. Salah satu strateginya adalah membuat aliran untuk setiap saluran target di AWS IoT Analytics atau aliran data Kinesis, meskipun Anda dapat menentukan beberapa target untuk aliran. Aliran memiliki masa pakai yang tahan lama.
### Persyaratan
Komponen ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.10.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.0
**catatan**
Membuat stream dengan tujuan ekspor Amazon S3 AWS IoT SiteWise atau Amazon memiliki persyaratan sebagai berikut:
Versi AWS IoT Greengrass Inti Minimum: 1.11.0
Versi SDK AWS IoT Greengrass Inti Minimum: Python: 1.6.0 \$1 Java: 1.5.0 \$1 Node.js: 1.7.0
### Contoh
Potongan berikut menciptakan aliran bernama `StreamName`. Potongan ini menentukan sifat aliran di `MessageStreamDefinition` dan jenis data bawahan.
------
#### [ 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.
```
[Referensi Python SDK: create\$1message\$1stream \$1 [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.create_message_stream)
------
#### [ 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.
}
```
Referensi 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.
});
```
Referensi Node.js SDK: \$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)
------
Untuk informasi lebih lanjut tentang konfigurasi tujuan ekspor, lihat[Ekspor konfigurasi untuk tujuan yang didukung AWS Cloud](stream-export-configurations.md).
## Tambahkan pesan
Untuk mengirim data ke pengelola pesan untuk ekspor, fungsi Lambda Anda menambahkan data ke pengaliran target. Tujuan ekspor menentukan tipe data yang lolos untuk metode ini.
### Persyaratan
Komponen ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.10.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.0
**catatan**
Menambahkan pesan dengan tujuan ekspor Amazon S3 AWS IoT SiteWise atau Amazon memiliki persyaratan berikut:
Versi AWS IoT Greengrass Inti Minimum: 1.11.0
Versi SDK AWS IoT Greengrass Inti Minimum: Python: 1.6.0 \$1 Java: 1.5.0 \$1 Node.js: 1.7.0
### Contoh
#### AWS IoT Analytics atau tujuan ekspor Kinesis Data Streams
Potongan berikut menambahkan pesan ke aliran bernama `StreamName`. Untuk AWS IoT Analytics atau tujuan Kinesis Data Streams, fungsi Lambda Anda menambahkan gumpalan data.
Potongan ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.10.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: 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.
```
Referensi 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.
}
```
Referensi 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.
});
```
Referensi SDK Node.js: [appendMessage](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage)
------
#### AWS IoT SiteWise tujuan ekspor
Potongan berikut menambahkan pesan ke aliran bernama `StreamName`. Untuk AWS IoT SiteWise tujuan, fungsi Lambda Anda menambahkan objek serial. `PutAssetPropertyValueEntry` Untuk informasi selengkapnya, lihat [Mengekspor ke AWS IoT SiteWise](stream-export-configurations.md#export-streams-to-sitewise).
**catatan**
Saat Anda mengirim data ke AWS IoT SiteWise, data Anda harus memenuhi persyaratan `BatchPutAssetPropertyValue` tindakan. Untuk informasi selengkapnya, lihat [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) di dalam *Referensi API AWS IoT SiteWise *.
Potongan ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.11.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: 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.
```
[Referensi Python SDK: 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.
}
```
Referensi Java SDK: [AppendMessage](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) \$1 [PutAssetPropertyValueEntry](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/sitewise/PutAssetPropertyValueEntry.html)
------
#### [ 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.
});
```
Referensi Node.js SDK: [AppendMessage](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) \$1 [PutAssetPropertyValueEntry](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.PutAssetPropertyValueEntry.html)
------
#### Tujuan ekspor Amazon S3
Cuplikan berikut menambahkan tugas ekspor ke pengaliran bernama `StreamName`. Untuk tujuan Amazon S3, fungsi Lambda Anda menambahkan `S3ExportTaskDefinition` objek berseri yang berisi informasi tentang file input sumber dan objek Amazon S3 target. Jika objek yang ditentukan tidak ada, Pengelola Pengaliran membuatnya untuk Anda. Untuk informasi selengkapnya, lihat [Mengekspor ke Amazon S3](stream-export-configurations.md#export-streams-to-s3).
Potongan ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.11.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: 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.
```
[https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.append_message](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()) {
// 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.
}
```
[Referensi Java SDK: [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.
});
```
[Referensi Node.js SDK: [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)
------
## Baca pesan
Baca pesan dari suatu aliran.
### Persyaratan
Komponen ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.10.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.0
### Contoh
Potongan berikut membaca pesan dari aliran bernama `StreamName`. Metode membaca mengambil objek `ReadMessagesOptions` opsional yang menentukan urutan nomor untuk memulai membaca, jumlah minimum dan maksimum yang dibaca, dan batas waktu untuk membaca pesan.
------
#### [ 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.
```
[Referensi Python SDK: 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.
}
```
Referensi 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.
});
```
Referensi Node.js SDK: [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)
------
## Daftar aliran
Dapatkan daftar aliran di stream manager.
### Persyaratan
Komponen ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.10.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.0
### Contoh
Potongan berikut mendapat daftar aliran (dengan nama) di stream manager.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
stream_names = client.list_streams()
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
Referensi 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.
}
```
Referensi 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.
});
```
Referensi SDK Node.js: [listStreams](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#listStreams)
------
## Jelaskan aliran pesan
Dapatkan metadata tentang aliran, termasuk definisi, ukuran, dan status ekspor aliran.
### Persyaratan
Komponen ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.10.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.0
### Contoh
Potongan berikut mendapat metadata tentang aliran bernama `StreamName`, termasuk status definisi, ukuran, dan pengekspor aliran.
------
#### [ 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.
```
Referensi 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.
}
```
Referensi Java SDK: [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.
});
```
Referensi Node.js SDK: [describeMessageStream](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#describeMessageStream)
------
## Perbarui aliran pesan
Perbarui properti dari aliran yang ada. Anda mungkin ingin memperbarui aliran jika kebutuhan Anda berubah setelah aliran dibuat. Contoh:
+ Tambahkan [konfigurasi ekspor](stream-export-configurations.md) baru untuk AWS Cloud tujuan.
+ Tingkatkan ukuran maksimum aliran untuk mengubah cara data diekspor atau disimpan. Misalnya, ukuran aliran yang dikombinasikan dengan strategi Anda pada pengaturan penuh dapat mengakibatkan data dihapus atau ditolak sebelum stream manager dapat memprosesnya.
+ Jeda dan lanjutkan ekspor; sebagai contoh, jika tugas ekspor berjalan lama dan Anda ingin menjatah data unggahan Anda.
Fungsi Lambda Anda mengikuti proses tingkat tinggi ini untuk memperbarui pengaliran:
1. [Dapatkan deskripsi aliran.](#streammanagerclient-describe-message-stream)
1. Perbarui properti target pada `MessageStreamDefinition` dan objek bawahan yang sesuai.
1. Lewati `MessageStreamDefinition` yang diperbarui. Pastikan untuk menyertakan definisi objek lengkap untuk aliran yang diperbarui. Properti yang tidak terdefinisi kembali ke nilai default.
Anda dapat menentukan nomor urutan pesan yang akan digunakan sebagai pesan awal dalam ekspor.
### Persyaratan
Komponen ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.11.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: Python: 1.6.0 \$1 Java: 1.5.0 \$1 Node.js: 1.7.0
### Contoh
Potongan berikut menciptakan aliran bernama `StreamName`. Potongan ini memperbarui beberapa properti aliran yang mengekspor ke 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.
```
Referensi Python SDK: \$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.
}
```
Referensi 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.
});
```
Referensi Node.js SDK: \$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)
------
### Kendala untuk memperbarui aliran
Kendala berikut berlaku saat memperbarui aliran. Kecuali tercantum dalam daftar berikut, pembaruan berlaku segera.
+ Anda tidak dapat memperbarui kegigihan aliran. Untuk mengubah perilaku ini, [hapus aliran](#streammanagerclient-delete-message-stream) dan [buat Stream](#streammanagerclient-create-message-stream) yang menentukan kebijakan kegigihan baru.
+ Anda dapat memperbarui ukuran maksimum aliran hanya dalam kondisi berikut:
+ Ukuran maksimum harus lebih besar atau sama dengan ukuran aliran. Untuk menemukan informasi ini, [jelaskan aliran](#streammanagerclient-describe-message-stream) dan kemudian periksa status penyimpanan `MessageStreamInfo` objek yang dikembalikan.
+ Ukuran maksimum harus lebih besar atau sama dengan ukuran segmen aliran.
+ Anda dapat memperbarui ukuran segmen aliran ke nilai kurang dari ukuran maksimum aliran tersebut. Pengaturan yang diperbarui berlaku untuk segmen baru.
+ Pembaruan ke properti waktu untuk tayang (TTL) berlaku untuk operasi append yang baru. Jika Anda mengurangi nilai ini, stream manager juga dapat menghapus segmen yang ada yang melebihi TTL.
+ Pembaruan untuk strategi tersebut pada properti penuh berlaku untuk operasi append yang baru. Jika Anda menetapkan strategi untuk menimpa data tertua, stream manager juga dapat menimpa segmen yang ada berdasarkan pengaturan yang baru.
+ Pembaruan untuk properti flush on write berlaku untuk pesan baru.
+ Pembaruan untuk konfigurasi ekspor berlaku untuk ekspor baru. Permintaan pembaruan harus mencakup semua konfigurasi ekspor yang ingin Anda dukung. Jika tidak, stream manager akan menghapusnya.
+ Saat Anda memperbarui konfigurasi ekspor, tentukan pengenal konfigurasi ekspor target.
+ Untuk menambahkan konfigurasi ekspor, tentukan pengenal unik untuk konfigurasi ekspor baru.
+ Untuk menghapus konfigurasi ekspor, hapus konfigurasi ekspor.
+ Untuk [memperbarui](#streammanagerclient-update-message-stream) nomor urutan awal konfigurasi ekspor di suatu aliran, Anda harus menentukan nilai yang kurang dari nomor urut terbaru. Untuk menemukan informasi ini, [jelaskan aliran](#streammanagerclient-describe-message-stream) dan kemudian periksa status penyimpanan `MessageStreamInfo` objek yang dikembalikan.
## Hapus aliran pesan
Hapus aliran. Bila Anda menghapus suatu aliran, semua data yang tersimpan untuk aliran itu akan dihapus dari disk.
### Persyaratan
Komponen ini memiliki persyaratan sebagai berikut:
+ Versi AWS IoT Greengrass Inti Minimum: 1.10.0
+ Versi SDK AWS IoT Greengrass Inti Minimum: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.0
### Contoh
Potongan berikut menghapus aliran bernama `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.
```
Referensi Python SDK: [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.
}
```
Referensi 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.
});
```
Referensi Node.js SDK: [deleteMessageStream](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#deleteMessageStream)
------
## Lihat juga
+ [Mengelola aliran data pada inti AWS IoT Greengrass](stream-manager.md)
+ [Konfigurasikan pengelola AWS IoT Greengrass aliran](configure-stream-manager.md)
+ [Ekspor konfigurasi untuk tujuan yang didukung AWS Cloud](stream-export-configurations.md)
+ [Ekspor aliran data ke AWS Cloud (konsol)](stream-manager-console.md)
+ [Ekspor aliran data ke AWS Cloud (CLI)](stream-manager-cli.md)
+ `StreamManagerClient`dalam referensi AWS IoT Greengrass Core SDK:
+ [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)