{"id":5281,"date":"2024-10-15T12:31:27","date_gmt":"2024-10-15T12:31:27","guid":{"rendered":"https:\/\/codethataint.com\/blog\/?p=5281"},"modified":"2024-10-15T13:01:55","modified_gmt":"2024-10-15T13:01:55","slug":"simple-subscriber-publisher-using-reactive-streams","status":"publish","type":"post","link":"https:\/\/codethataint.com\/blog\/simple-subscriber-publisher-using-reactive-streams\/","title":{"rendered":"Simple Subscriber Publisher Using Reactive Streams"},"content":{"rendered":"
    \n
  1. publisher does not produce data unless subscriber requests for it.<\/li>\n
  2. publisher will produce only <= subscriber requested items. publisher can also produce 0 items!<\/li>\n
  3. subscriber can cancel the subscription. producer should stop at that moment as subscriber is no longer interested in consuming the data<\/li>\n
  4. producer can send the error signal<\/li>\n<\/ol>\n

    \"\"<\/p>\n

      \n
    1. PublisherImpl<\/strong> instance would call the subscribe method and pass instance of SubscriberImpl<\/strong><\/li>\n
    2. PublisherImpl<\/strong> subscribe method would inturn call the onSubscribe method using the instance of SubscriberImpl<\/strong> passed.<\/li>\n
    3. SubscriberImpl<\/strong> would get the same subscription which it has been passed to publisherImpl<\/strong> subscribe method earlier<\/li>\n
    4. PublisherImpl <\/strong>has following methods\n
        \n
      • subscribe – takes subscriber as argument and creates new subscription and notify the subscriber by calling the onSubscribe method\n
        \r\npublic void subscribe(Subscriber subscriber) {\r\n        var subscription = new SubscriptionImpl(subscriber);\r\n        subscriber.onSubscribe(subscription);\r\n}\r\n<\/pre>\n<\/li>\n<\/ul>\n
      • SubscriberImpl <\/strong>has following methods\n
          \n
        • onSubscribe – To get the same subscription passed to publisher subscribe method. This is inturn called from publisherImpl<\/li>\n
        • onNext – called from subscriptionImpl to pass the data during iteration<\/li>\n
        • onError – called from subscriptionImpl during error<\/li>\n
        • onComplete – called from when all data is completed<\/li>\n<\/ul>\n
          \r\n@Override\r\n    public void onSubscribe(Subscription subscription) {\r\n        this.subscription = subscription;\r\n    }\r\n\r\n    @Override\r\n    public void onNext(String emailId) {\r\n        logger.info("Received {}", emailId);\r\n    }\r\n\r\n    @Override\r\n    public void onError(Throwable throwable) {\r\n        logger.info("---------------------------------------------");\r\n        logger.info("Received error {}", throwable.getMessage());\r\n    }\r\n\r\n    @Override\r\n    public void onComplete() {\r\n        logger.info("Subscription ended");\r\n    }\r\n<\/pre>\n<\/li>\n
        • SubscriptionImpl<\/strong>has following methods\n
            \n
          • request – you can request date using subscriptionImpl instance by passing no of records<\/li>\n
          • cancel – cancel subscription<\/li>\n<\/ul>\n
            \r\n@Override\r\n    public void request(long requestedItemCnt) {\r\n        if(isCancelled){\r\n            return;\r\n        }\r\n\r\n        logger.info("Subscriber has requested {} items ", requestedItemCnt);\r\n\r\n        if(requestedItemCnt >MAX_ITEMS){\r\n            this.subscriber.onError(new RuntimeException(" Items requested is more than Total Items Available"));\r\n            this.isCancelled = true;\r\n            return;\r\n        }\r\n\r\n        \/\/Check if all items(MAX_ITEMS) were sent\r\n        for(int idx=0;idx<requestedItemCnt && count<MAX_ITEMS; idx++){\r\n            count++;\r\n            this.subscriber.onNext(this.faker.internet().emailAddress());\r\n        }\r\n\r\n        \/\/If all items were sent complete subscription\r\n        if(count == MAX_ITEMS){\r\n            logger.info("No More data from Producer");\r\n            this.subscriber.onComplete();\r\n            isCancelled = true;\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public void cancel() {\r\n        logger.info("Cancelling Subscription... . . .");\r\n        isCancelled = true;\r\n    }\r\n<\/pre>\n<\/li>\n<\/ol>\n
            pom.xml<\/strong><\/p>\n
            \r\n  <dependency>\r\n            <groupId>io.projectreactor<\/groupId>\r\n            <artifactId>reactor-core<\/artifactId>\r\n        <\/dependency>\r\n        <dependency>\r\n            <groupId>io.projectreactor.netty<\/groupId>\r\n            <artifactId>reactor-netty-core<\/artifactId>\r\n        <\/dependency>\r\n        <dependency>\r\n            <groupId>io.projectreactor.netty<\/groupId>\r\n            <artifactId>reactor-netty-http<\/artifactId>\r\n        <\/dependency>\r\n        <dependency>\r\n            <groupId>ch.qos.logback<\/groupId>\r\n            <artifactId>logback-classic<\/artifactId>\r\n            <version>${logback.version}<\/version>\r\n        <\/dependency>\r\n        <dependency>\r\n            <groupId>com.github.javafaker<\/groupId>\r\n            <artifactId>javafaker<\/artifactId>\r\n            <version>${faker.version}<\/version>\r\n        <\/dependency>\r\n<\/pre>\n
            PublisherImpl.java<\/strong><\/p>\n
            \r\nimport org.reactivestreams.Publisher;\r\nimport org.reactivestreams.Subscriber;\r\n\r\npublic class PublisherImpl implements Publisher {\r\n    @Override\r\n    public void subscribe(Subscriber subscriber) {\r\n        var subscription = new SubscriptionImpl(subscriber);\r\n        subscriber.onSubscribe(subscription);\r\n    }\r\n}\r\n<\/pre>\n
            SubscriptionImpl.java<\/strong><\/p>\n
            \r\npublic class SubscriptionImpl implements Subscription {\r\n    private static final Logger logger = LoggerFactory.getLogger(SubscriptionImpl.class);\r\n    private final Subscriber<? super String> subscriber;\r\n    private boolean isCancelled = false;\r\n    private final Faker faker;\r\n\r\n    private final int MAX_ITEMS = 10;\r\n    private static int count = 0;\r\n\r\n\r\n    public SubscriptionImpl(Subscriber subscriber){\r\n        this.subscriber = subscriber;\r\n        this.faker = Faker.instance();\r\n    }\r\n\r\n    @Override\r\n    public void request(long requestedItemCnt) {\r\n        if(isCancelled){\r\n            return;\r\n        }\r\n\r\n        logger.info("Subscriber has requested {} items ", requestedItemCnt);\r\n\r\n        if(requestedItemCnt >MAX_ITEMS){\r\n            this.subscriber.onError(new RuntimeException(" Items requested is more than Total Items Available"));\r\n            this.isCancelled = true;\r\n            return;\r\n        }\r\n\r\n        \/\/Check if all items(MAX_ITEMS) were sent\r\n        for(int idx=0;idx<requestedItemCnt && count<MAX_ITEMS; idx++){\r\n            count++;\r\n            this.subscriber.onNext(this.faker.internet().emailAddress());\r\n        }\r\n\r\n\r\n        \/\/If all items were sent complete subscription\r\n        if(count == MAX_ITEMS){\r\n            logger.info("No More data from Producer");\r\n            this.subscriber.onComplete();\r\n            isCancelled = true;\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public void cancel() {\r\n        logger.info("Cancelling Subscription... . . .");\r\n        isCancelled = true;\r\n    }\r\n}\r\n<\/pre>\n
            SubscriberImpl.java<\/strong><\/p>\n
            \r\nimport org.reactivestreams.Subscriber;\r\nimport org.reactivestreams.Subscription;\r\nimport org.slf4j.Logger;\r\nimport org.slf4j.LoggerFactory;\r\n\r\npublic class SubscriberImpl implements Subscriber<String> {\r\n    private static final Logger logger = LoggerFactory.getLogger(SubscriberImpl.class);\r\n    private Subscription subscription;\r\n\r\n    @Override\r\n    public void onSubscribe(Subscription subscription) {\r\n        this.subscription = subscription;\r\n    }\r\n\r\n    @Override\r\n    public void onNext(String emailId) {\r\n        logger.info("Received {}", emailId);\r\n    }\r\n\r\n    @Override\r\n    public void onError(Throwable throwable) {\r\n        logger.info("---------------------------------------------");\r\n        logger.info("Received error {}", throwable.getMessage());\r\n    }\r\n\r\n    @Override\r\n    public void onComplete() {\r\n        logger.info("Subscription ended");\r\n    }\r\n\r\n    public Subscription getSubscription() {\r\n        return subscription;\r\n    }\r\n}\r\n<\/pre>\n
            Main.java<\/strong><\/p>\n
            \r\nimport org.mugil.publisher.PublisherImpl;\r\nimport org.mugil.subscriber.SubscriberImpl;\r\n\r\nimport java.time.Duration;\r\n\r\npublic class Main {\r\n    public static void main(String[] args) throws InterruptedException {\r\n        getMessages();\r\n    }\r\n\r\n    public static void getMessages() throws InterruptedException {\r\n        var objPublisher = new PublisherImpl();\r\n        var objSubscriber = new SubscriberImpl();\r\n\r\n        objPublisher.subscribe(objSubscriber);\r\n        objSubscriber.getSubscription().request(1);\r\n        Thread.sleep(Duration.ofSeconds(2));\r\n\r\n        objSubscriber.getSubscription().request(2);\r\n        Thread.sleep(Duration.ofSeconds(2));\r\n\r\n        objSubscriber.getSubscription().request(3);\r\n        Thread.sleep(Duration.ofSeconds(2));\r\n\r\n        objSubscriber.getSubscription().cancel();\r\n\r\n        objSubscriber.getSubscription().request(1);\r\n    }\r\n}\r\n<\/pre>\n
            Output<\/strong><\/p>\n
            \r\n18:00:57.240 [main] INFO org.mugil.publisher.SubscriptionImpl -- Subscriber has requested 1 items \r\n18:00:57.534 [main] INFO org.mugil.subscriber.SubscriberImpl -- Received jani.mante@gmail.com\r\n18:00:59.537 [main] INFO org.mugil.publisher.SubscriptionImpl -- Subscriber has requested 2 items \r\n18:00:59.541 [main] INFO org.mugil.subscriber.SubscriberImpl -- Received sunny.quigley@yahoo.com\r\n18:00:59.544 [main] INFO org.mugil.subscriber.SubscriberImpl -- Received hang.gutkowski@yahoo.com\r\n18:01:01.546 [main] INFO org.mugil.publisher.SubscriptionImpl -- Subscriber has requested 3 items \r\n18:01:01.548 [main] INFO org.mugil.subscriber.SubscriberImpl -- Received malik.thiel@hotmail.com\r\n18:01:01.549 [main] INFO org.mugil.subscriber.SubscriberImpl -- Received andre.purdy@gmail.com\r\n18:01:01.550 [main] INFO org.mugil.subscriber.SubscriberImpl -- Received kim.greenfelder@gmail.com\r\n18:01:03.560 [main] INFO org.mugil.publisher.SubscriptionImpl -- Cancelling Subscription... . . .\r\n<\/pre>\n","protected":false},"excerpt":{"rendered":"
            publisher does not produce data unless subscriber requests for it. publisher will produce only<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[369],"tags":[],"class_list":["post-5281","post","type-post","status-publish","format-standard","hentry","category-reactive-programming"],"_links":{"self":[{"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/posts\/5281","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/comments?post=5281"}],"version-history":[{"count":6,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/posts\/5281\/revisions"}],"predecessor-version":[{"id":5290,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/posts\/5281\/revisions\/5290"}],"wp:attachment":[{"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/media?parent=5281"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/categories?post=5281"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/tags?post=5281"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}