Create and Manage Subscriptions

Attributes of a Subscription

  • ackTimeoutSeconds.

    Specifies the max amount of time for the server to mark the record as unacknowledged, after which the record will be sent again.

  • maxUnackedRecords.

    The maximum amount of unacknowledged records allowed. After exceeding the size set, the server will stop sending records to corresponding consumers.

Create a subscription

Every subscription has to specify which stream to subscribe to, which means you have to make sure the stream to be subscribed has already been created.

For the subscription name, please refer to the guidelines to name a resource

When creating a subscription, you can provide the attributes mentioned like this:

  1. // CreateSubscriptionExample.java
  3. package docs.code.examples;
  5. import io.hstream.HStreamClient;
  6. import io.hstream.Subscription;
  8. public class CreateSubscriptionExample {
  9.   public static void main(String[] args) throws Exception {
  10.     // TODO(developer): Replace these variables before running the sample.
  11.     String serviceUrl = "127.0.0.1:6570";
  12.     if (System.getenv("serviceUrl") != null) {
  13.       serviceUrl = System.getenv("serviceUrl");
  14.     }
  15.     String streamName = "your_h_records_stream_name";
  16.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  17.     createSubscriptionExample(client, streamName);
  18.     client.close();
  19.   }
  21.   public static void createSubscriptionExample(HStreamClient client, String streamName) {
  22.     // TODO(developer): Specify the options while creating the subscription
  23.     String subscriptionId = "your_subscription_id";
  24.     Subscription subscription =
  25.         Subscription.newBuilder().subscription(subscriptionId).stream(streamName)
  26.             .ackTimeoutSeconds(600) // The default setting is 600 seconds
  27.             .maxUnackedRecords(10000) // The default setting is 10000 records
  28.             .build();
  29.     client.createSubscription(subscription);
  30.   }
  31. }
  1. // ExampleCreateSubscription.go
  3. package examples
  5. import (
  6.     "github.com/hstreamdb/hstreamdb-go/hstream"
  7.     "log"
  8. )
  10. func ExampleCreateSubscription() 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.     streamName := "testStream"
  18.     subId0 := "SubscriptionId0"
  19.     subId1 := "SubscriptionId1"
  21.     // Create a new subscription with ack timeout = 60s, max unAcked records num set to 10000 and set
  22.     // subscriptionOffset to Earliest
  23.     if err = client.CreateSubscription(subId0, streamName,
  24.         hstream.WithAckTimeout(60),
  25.         hstream.WithMaxUnackedRecords(10000),
  26.         hstream.WithOffset(hstream.EARLIEST)); err != nil {
  27.         log.Fatalf("Creating subscription error: %s", err)
  28.     }
  30.     if err = client.CreateSubscription(subId1, streamName,
  31.         hstream.WithAckTimeout(600),
  32.         hstream.WithMaxUnackedRecords(5000),
  33.         hstream.WithOffset(hstream.LATEST)); err != nil {
  34.         log.Fatalf("Creating subscription error: %s", err)
  35.     }
  37.     return nil
  38. }
  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_subscription(client):
  21.     await client.create_subscription(
  22.         subscription,
  23.         stream_name,
  24.         ack_timeout=600,
  25.         max_unacks=10000,
  26.         offset=hstreamdb.SpecialOffset.EARLIEST,
  27.     )

Delete a subscription

To delete a subscription without the force flag, you need to make sure that there is no active subscription consumer.

Delete a subscription with the force flag

If you do want to delete a subscription with running consumers, enable force deletion. While force deleting a subscription, the subscription will be in deleting state and closing running consumers, which means you will not be able to join, delete or create a subscription with the same name. After the deletion completes, you can create a subscription with the same name. However, this new subscription will be a brand new subscription. Even if they subscribe to the same stream, this new subscription will not share the consumption progress with the deleted subscription.

  1. // DeleteSubscriptionExample.java
  3. package docs.code.examples;
  5. import io.hstream.HStreamClient;
  7. public class DeleteSubscriptionExample {
  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 subscriptionId = "your_subscription_id";
  17.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  18.     deleteSubscriptionExample(client, subscriptionId);
  19.     client.close();
  20.   }
  22.   public static void deleteSubscriptionExample(HStreamClient client, String subscriptionId) {
  23.     client.deleteSubscription(subscriptionId);
  24.   }
  26.   public static void deleteSubscriptionForceExample(HStreamClient client, String subscriptionId) {
  27.     client.deleteSubscription(subscriptionId, true);
  28.   }
  29. }
  1. // ExampleDeleteSubscription.go
  3. package examples
  5. import (
  6.     "github.com/hstreamdb/hstreamdb-go/hstream"
  7.     "log"
  8. )
  10. func ExampleDeleteSubscription() 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.     subId0 := "SubscriptionId0"
  18.     subId1 := "SubscriptionId1"
  20.     // force delete subscription
  21.     if err = client.DeleteSubscription(subId0, true); err != nil {
  22.         log.Fatalf("Force deleting subscription error: %s", err)
  23.     }
  25.     // delete subscription
  26.     if err = client.DeleteSubscription(subId1, false); err != nil {
  27.         log.Fatalf("Deleting subscription error: %s", err)
  28.     }
  30.     return nil
  31. }
  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_subscription(client):
  21.     await client.delete_subscription(subscription, force=True)

List subscriptions

To list all subscriptions in HStream

  1. // ListSubscriptionsExample.java
  3. package docs.code.examples;
  5. import io.hstream.HStreamClient;
  6. import io.hstream.Subscription;
  7. import java.util.List;
  9. public class ListSubscriptionsExample {
  11.   public static void main(String[] args) throws Exception {
  12.     // TODO(developer): Replace these variables before running the sample.
  13.     String serviceUrl = "127.0.0.1:6570";
  14.     if (System.getenv("serviceUrl") != null) {
  15.       serviceUrl = System.getenv("serviceUrl");
  16.     }
  18.     HStreamClient client = HStreamClient.builder().serviceUrl(serviceUrl).build();
  19.     listSubscriptionExample(client);
  20.     client.close();
  21.   }
  23.   public static void listSubscriptionExample(HStreamClient client) {
  24.     List<Subscription> subscriptions = client.listSubscriptions();
  25.     for (Subscription subscription : subscriptions) {
  26.       System.out.println(subscription.getSubscriptionId());
  27.     }
  28.   }
  29. }
  1. // ExampleListSubscriptions.go
  3. package examples
  5. import (
  6.     "fmt"
  7.     "github.com/hstreamdb/hstreamdb-go/hstream"
  8.     "log"
  9. )
  11. func ExampleListSubscriptions() 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.     subscriptions, err := client.ListSubscriptions()
  19.     if err != nil {
  20.         log.Fatalf("Listing subscriptions error: %s", err)
  21.     }
  23.     for _, sub := range subscriptions {
  24.         fmt.Printf("%+v\n", sub)
  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_subscriptions(client):
  21.     subscriptions = await client.list_subscriptions()
  22.     for s in subscriptions:
  23.         print(s)