The Subscriber verification

The mentioned configurations are the minimum that are required in order to start testing our producer's behavior. However, along with the instances of  Publisher, we have instances of  Subscriber that should be tested as well. Fortunately, that group of rules in the Reactive Stream specification is less complex than Publisher's one, but it is still required to satisfy all requirements.

There are two different test suits to test NewsServiceSubscriber. The first one is called org.reactivestreams.tck.SubscriberBlackboxVerification, which allows verifying the Subscriber without knowledge or modification of its internals. Blackbox verification is a useful test kit when the Subscriber comes from the external codebase, and there is no legal way to extend the behavior. On the other hand, the Blackbox verification covers only a few rules and does not ensure complete correctness of the implementation. To see how the NewsServiceSubscriber may be examined, let’s implement the Blackbox verification test first:

public class NewsServiceSubscriberTest                             // (1)    
   extends SubscriberBlackboxVerification<NewsLetter> {            // 

   public NewsServiceSubscriberTest() {                            // (2)
      super(new TestEnvironment());                                //
   }                                                               //

   @Override                                                       // (3)
   public Subscriber<NewsLetter> createSubscriber() {              //
      return new NewsServiceSubscriber(...);                       //
   }                                                               //

   @Override                                                       // (4)
   public NewsLetter createElement(int element) {                  //
      return new StubNewsLetter(element);                          //
   }                                                               //

   @Override                                                       // (5)
   public void triggerRequest(Subscriber<? super NewsLetter> s) {  //
      ((NewsServiceSubscriber) s).eventuallyReadDigest();          // (5.1)
   }                                                               //
}

The key is as follows:

1. This is the NewsServiceSubscriberTest class declaration which extends the SubscriberBlackboxVerification tests-suite.
2. This is the default constructor declaration. Here, identically to PublisherVerification, we are mandated to provide a certain TestEnvironment.
3. This is the createSubscriber method implementation. Here, that method returns the NewsServiceSubscriber instance, which should be tested against the specification.

4. This is the createElement method implementation. Here, we are required to provide an implementation of the method which plays the role of a new element factory and generates a new instance of NewsLetter on demand.
5. This is the triggerRequest method implementation. Since the Blackbox testing assumes no access to the internals, it means that we do not have direct access to the hidden Subscription inside the Subscriber. Subsequently, this means that we have to trigger it somehow by manually using the given API (5.1).

The preceding example shows the available API for Subscriber verification. Apart from the two required methods, createSubscriber and createElement, there is an additional method which addresses the handling of the Subscription#request method externally. In our case, it is a useful addition that allows us to emulate real user activity.

The second test kit is called org.reactivestreams.tck.SubscriberWhiteboxVerification. This is a similar verification to the previous one, but to pass the verification, the Subscriber should provide additional interaction with the WhiteboxSubscriberProbe :

public class NewsServiceSubscriberWhiteboxTest                     // (1)
   extends SubscriberWhiteboxVerification<NewsLetter> {            //
   ...                                                             //

   @Override                                                       // (2)
   public Subscriber<NewsLetter> createSubscriber(                 //
      WhiteboxSubscriberProbe<NewsLetter> probe                    //
   ) {                                                             //
      return new NewsServiceSubscriber(...) {                      //
         public void onSubscribe(Subscription s) {                 //
            super.onSubscribe(s);                                  // (2.1)
            probe.registerOnSubscribe(new SubscriberPuppet() {     // (2.2)
               public void triggerRequest(long elements) {         //
                  s.request(elements);                             //
               }                                                   //
               public void signalCancel() {                        //
                  s.cancel();                                      //
               }                                                   //
            });                                                    //
         }                                                         //
         public void onNext(NewsLetter newsLetter) {               //
            super.onNext(newsLetter);                              //
            probe.registerOnNext(newsLetter);                      // (2.3)
         }                                                         //
         public void onError(Throwable t) {                        //
            super.onError(t);                                      //
            probe.registerOnError(t);                              // (2.4)
         }                                                         //
         public void onComplete() {                                //
            super.onComplete();                                    //
            probe.registerOnComplete();                            // (2.5)
         }                                                         //
      };                                                           //
   }                                                               //
   ...                                                             //
}                                                                  //

The key is as follows:

1. This is the NewsServiceSubscriberWhiteboxTest class declaration which extends the SubscriberWhiteboxVerification tests-suite.
2. This is the createSubscriber method implementation. This method works identically to the Blackbox verification and returns the Subscriber instance, but here there is an additional parameter called WhiteboxSubscriberProbe. In that case, WhiteboxSubscriberProbe represents a mechanism that enables embedded control of the demand and capture of the incoming signals. In comparison to the Blackbox verification, by proper registration of probe hooks inside NewsServiceSubscriber, (2.2), (2.3), (2.4), (2.5), the test suite is capable not only of sending the demand but verifying that the demand was satisfied and all elements have been received as well. In turn, the mechanism of demand regulation is more transparent than it previously was. Here, at point (2.2), we implement the SubscriberPuppet, which adapts direct access to the received Subscription. 

As we can see, opposite to the Blackbox verification, the Whitebox requires the extension of the Subscriber, providing additional hooks internally. While the Whitebox testing covers a broader number of rules which ensure the correct behavior of the tested Subscriber, it may be unacceptable for those cases when we want to make a class final to prevent it from being extended.

The final part of the verification journey is the testing of the Processor. For that purpose, TCK provides us with org.reactivestreams.tck.IdentityProcessorVerification. This test suite can verify a Processor, which receives and produces the same type of elements. In our example, only the martMulticastProcessor behaves in such a manner. Since the test kit should verify the behavior of both Publisher and Subscriber, the IdentityProcessorVerification inherits similar configurations as for the Publisher and Subscriber tests. Consequently, we do not get into the details of the whole test's implementation, but consider additional methods required for the SmartMulticastProcessor verification:

public class SmartMulticastProcessorTest                           // (1)
   extends IdentityProcessorVerification<NewsLetter> {             //

   public SmartMulticastProcessorTest() {                          // (2)
      super(..., 1);                                               //
   }                                                               //

   @Override                                                       // (3)
   public Processor<Integer, Integer> createIdentityProcessor(     //
      int bufferSize                                               //
   ) {                                                             //
      return new SmartMulticastProcessor<>();                      //
   }                                                               //

   @Override                                                       // (4)
   public NewsLetter createElement(int element) {                  //
      return new StubNewsLetter(element);                          //
   }                                                               //
}

The key is as follows:

1. This is the SmartMulticastProcessorTest class definition, which extends IdentityProcessorVerification.
2. This is the default constructor definition. As we may notice from the code, (along with the TestEnvironment configuration, which is skipped in that example) we pass an additional parameter, which indicates the number of elements that the processor must buffer without dropping. Since we know that our Processor supports the buffering of only one element, we have to provide that number manually before starting any verification.

3. This is the createIdentityProcessor method implementation, which returns an instance of the tested Processor. Here, the bufferSize represents the number of elements that the Processor must buffer without dropping. We may skip that parameter now, since we know that the internal buffer size is equal to the pre-configured one in the constructor.
4. This is the createElement method implementation. Similar to the verification of Subscriber, we have to provide the factory method to create new elements. 

The preceding example shows the essential configuration for SmartMulticastProcessor verification. Since the IdentityProcessorVerification extends both SubscriberWhiteboxVerification and PublisherVerification, the general configurations are merged from each of them.

To generalize, we got an overview of the essential set of tests that help to verify the specified behavior of the implemented Reactive Operators. Here, TCK may be considered as initial integration tests. Nevertheless, we should bear in mind that along with the TCK verification, each operator should carefully test for the desired behavior on its own.