Create and Manage Streams

Guidelines to name a resource

An HStream resource name uniquely identifies an HStream resource, such as a stream, a subscription or a reader. The resource name must fit the following requirements:

  • Start with a letter
  • Length must be no longer than 255 characters
  • Contain only the following characters: Letters [A-Za-z], numbers [0-9], dashes -, underscores _

*For the cases where the resource name is used as a part of a SQL statement, such as in HStream SQL Shell, there will be situations where the resource name cannot be parsed properly (such as conflicts with Keywords etc.), enclose the resource name with backticks ` . With the enhancements of the SQL parser, the constriction may be removed in the future.

Attributes of a Stream

  • Replication factor

    For fault tolerance and higher availability, every stream can be replicated across nodes in the cluster. A typical production setting is a replication factor of 3, i.e., there will always be three copies of your data, which is helpful just in case things go wrong or you want to do maintenance on the brokers. This replication is performed at the level of the stream.

  • Backlog retention

    The configuration controls how long streams of HStreamDB retain records after being appended. HStreamDB will discard the message regardless of whether it is consumed when it exceeds the backlog retention duration.

    • Default = 7 days
    • Minimum value = 1 seconds
    • Maximum value = 21 days

  • Shard Count

    The number of shards that a stream will have.

Create a Stream

Create a stream before you write records or create a subscription.

  1. // CreateStreamExample.java
  3. package docs.code.examples;
  5. import io.hstream.HStreamClient;
  7. public class CreateStreamExample {
  8.   public static void main(String[] args) throws Exception {
  9.     // TODO(developer): Replace these variables before running the sample.
  10.     String serviceUrl = "127.0.0.1:6570";
  11.     if (System.getenv("serviceUrl") != null) {
  12.       serviceUrl = System.getenv("serviceUrl");
  13.     }
  15.     String streamName1 = "your_h_records_stream_name";
  16.     String streamName2 = "your_raw_records_stream_name";
  18.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  19.     createStreamExample(client, streamName1);
  20.     createStreamWithAttrsExample(client, streamName2);
  21.     client.close();
  22.   }
  24.   public static void createStreamExample(HStreamClient client, String streamName) {
  25.     client.createStream(streamName);
  26.   }
  28.   public static void createStreamWithAttrsExample(HStreamClient client, String streamName) {
  29.     client.createStream(
  30.         streamName,
  31.         (short) 1 // replication factor
  32.         ,
  33.         10 // Number of shards
  34.         ,
  35.         7 * 24 * 3600 // backlog retention time in seconds
  36.         );
  37.   }
  38. }
  1. // ExampleCreateStream.go
  3. package examples
  5. import (
  6.     "log"
  8.     "github.com/hstreamdb/hstreamdb-go/hstream"
  9. )
  11. func ExampleCreateStream() error {
  12.     client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
  13.     if err != nil {
  14.         log.Fatalf("Creating client error: %s", err)
  15.     }
  16.     defer client.Close()
  18.     // Create a stream, only specific streamName
  19.     if err = client.CreateStream("testDefaultStream"); err != nil {
  20.         log.Fatalf("Creating stream error: %s", err)
  21.     }
  23.     // Create a new stream with 1 replica, 5 shards, set the data retention to 1800s.
  24.     err = client.CreateStream("testStream",
  25.         hstream.WithReplicationFactor(1),
  26.         hstream.EnableBacklog(1800),
  27.         hstream.WithShardCount(5))
  28.     if err != nil {
  29.         log.Fatalf("Creating stream error: %s", err)
  30.     }
  32.     return nil
  33. }
  1. # https://github.com/hstreamdb/hstreamdb-py/blob/main/examples/snippets/guides.py
  2. import asyncio
  3. import hstreamdb
  4. import os
  6. # NOTE: Replace with your own host and port
  7. host = os.getenv("GUIDE_HOST", "127.0.0.1")
  8. port = os.getenv("GUIDE_PORT", 6570)
  9. stream_name = "your_stream"
  10. subscription = "your_subscription"
  13. # Run: asyncio.run(main(your_async_function))
  14. async def main(*funcs):
  15.     async with await hstreamdb.insecure_client(host=host, port=port) as client:
  16.         for f in funcs:
  17.             await f(client)
  20. async def create_stream(client):
  21.     await client.create_stream(
  22.         stream_name, replication_factor=1, backlog=24 * 60 * 60, shard_count=1
  23.     )

Delete a Stream

Deletion is only allowed when a stream has no subsequent subscriptions unless the force flag is set.

Delete a stream with the force flag

If you need to delete a stream with subscriptions, enable force deletion. Existing stream subscriptions can still read from the backlog after deleting a stream with the force flag enabled. However, these subscriptions will have stream name __deleted_stream__, no new subscription creation on the deleted stream would be allowed, nor new records would be allowed to be written to the stream.

  1. // DeleteStreamExample.java
  3. package docs.code.examples;
  5. import io.hstream.HStreamClient;
  7. public class DeleteStreamExample {
  8.   public static void main(String[] args) throws Exception {
  9.     // TODO(developer): Replace these variables before running the sample.
  10.     // String serviceUrl = "your-service-url-address";
  11.     String serviceUrl = "127.0.0.1:6570";
  12.     if (System.getenv("serviceUrl") != null) {
  13.       serviceUrl = System.getenv("serviceUrl");
  14.     }
  16.     String streamName1 = "your_h_records_stream_name";
  17.     String streamName2 = "your_raw_records_stream_name";
  19.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  20.     deleteStreamExample(client, streamName1);
  21.     deleteStreamForceExample(client, streamName2);
  22.     client.close();
  23.   }
  25.   public static void deleteStreamExample(HStreamClient client, String streamName) {
  26.     client.deleteStream(streamName);
  27.   }
  29.   public static void deleteStreamForceExample(HStreamClient client, String streamName) {
  30.     client.deleteStream(streamName, true);
  31.   }
  32. }
  1. // ExampleDeleteStream.go
  3. package examples
  5. import (
  6.     "github.com/hstreamdb/hstreamdb-go/hstream"
  7.     "log"
  8. )
  10. func ExampleDeleteStream() error {
  11.     client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
  12.     if err != nil {
  13.         log.Fatalf("Creating client error: %s", err)
  14.     }
  15.     defer client.Close()
  17.     // force delete stream and ignore none exist stream
  18.     if err := client.DeleteStream("testStream",
  19.         hstream.EnableForceDelete,
  20.         hstream.EnableIgnoreNoneExist); err != nil {
  21.         log.Fatalf("Deleting stream error: %s", err)
  22.     }
  24.     if err := client.DeleteStream("testDefaultStream"); err != nil {
  25.         log.Fatalf("Deleting stream error: %s", err)
  26.     }
  28.     return nil
  29. }
  1. # https://github.com/hstreamdb/hstreamdb-py/blob/main/examples/snippets/guides.py
  2. import asyncio
  3. import hstreamdb
  4. import os
  6. # NOTE: Replace with your own host and port
  7. host = os.getenv("GUIDE_HOST", "127.0.0.1")
  8. port = os.getenv("GUIDE_PORT", 6570)
  9. stream_name = "your_stream"
  10. subscription = "your_subscription"
  13. # Run: asyncio.run(main(your_async_function))
  14. async def main(*funcs):
  15.     async with await hstreamdb.insecure_client(host=host, port=port) as client:
  16.         for f in funcs:
  17.             await f(client)
  20. async def delete_stream(client):
  21.     await client.delete_stream(stream_name, ignore_non_exist=True, force=True)

List Streams

To get all streams in HStreamDB:

  1. // ListStreamsExample.java
  3. package docs.code.examples;
  5. import io.hstream.HStreamClient;
  6. import io.hstream.Stream;
  7. import java.util.List;
  9. public class ListStreamsExample {
  10.   public static void main(String[] args) throws Exception {
  11.     // TODO(developer): Replace these variables before running the sample.
  12.     String serviceUrl = "127.0.0.1:6570";
  13.     if (System.getenv("serviceUrl") != null) {
  14.       serviceUrl = System.getenv("serviceUrl");
  15.     }
  17.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  18.     listStreamExample(client);
  19.     client.close();
  20.   }
  22.   public static void listStreamExample(HStreamClient client) {
  23.     List<Stream> streams = client.listStreams();
  24.     for (Stream stream : streams) {
  25.       System.out.println(stream.getStreamName());
  26.     }
  27.   }
  28. }

s

  1. // ExampleListStreams.go
  3. package examples
  5. import (
  6.     "fmt"
  7.     "github.com/hstreamdb/hstreamdb-go/hstream"
  8.     "log"
  9. )
  11. func ExampleListStreams() error {
  12.     client, err := hstream.NewHStreamClient(YourHStreamServiceUrl)
  13.     if err != nil {
  14.         log.Fatalf("Creating client error: %s", err)
  15.     }
  16.     defer client.Close()
  18.     streams, err := client.ListStreams()
  19.     if err != nil {
  20.         log.Fatalf("Listing streams error: %s", err)
  21.     }
  23.     for _, stream := range streams {
  24.         fmt.Printf("%+v\n", stream)
  25.     }
  27.     return nil
  28. }
  1. # https://github.com/hstreamdb/hstreamdb-py/blob/main/examples/snippets/guides.py
  2. import asyncio
  3. import hstreamdb
  4. import os
  6. # NOTE: Replace with your own host and port
  7. host = os.getenv("GUIDE_HOST", "127.0.0.1")
  8. port = os.getenv("GUIDE_PORT", 6570)
  9. stream_name = "your_stream"
  10. subscription = "your_subscription"
  13. # Run: asyncio.run(main(your_async_function))
  14. async def main(*funcs):
  15.     async with await hstreamdb.insecure_client(host=host, port=port) as client:
  16.         for f in funcs:
  17.             await f(client)
  20. async def list_streams(client):
  21.     ss = await client.list_streams()
  22.     for s in ss:
  23.         print(s)