Previous Next JavaScript must be enabled to correctly display this content

## Using the NoSQLSubscriber Interface

You implement subscribers using the `oracle.kv.pubsub.NoSQLSubscriber` interface, an extension of `org.reactivestreams.Subscriber`. `NoSQLSubscriber` provides the following methods, which you must implement:

- `onSubscribe()`

  This is the method invoked after the publisher has successfully established contact with the Oracle NoSQL Database store. The argument you pass to this method is an `org.reactivestreams.Subscription` instance, which you can then cast to `oracle.kv.pubsub.NoSQLSubscription`. See Working with Subscriptions.

      ...
      private NoSQLSubscription subscription;
      ...

      @Override
      public void onSubscribe(Subscription s) {
          subscription = (NoSQLSubscription) s;
          // request 100 store operations be streamed to this
          // subscriber.
          s.request(100);
      }

  Note:

  You do not have to call the `s.request(100)`method inside `onSubscribe()`. Once an instance of `NoSQLSubscription` is available, you can call the method outside `onSubscribe()`. The main point of the `onSubscribe()` method here is to pass the user the subscription instance that the publisher generates.

- `onNext()`

  Signals the next Oracle NoSQL Database operation. You pass this method a `StreamOperation` class instance. See Using the StreamOperation Class. This method is where you perform whatever processing you want to perform on the stream events.

      @Override
      public void onNext(StreamOperation t) {
          // perform processing on the StreamOperation
          // here. Typically you will do different
          // things depending on whether this is
          // a put,delete, or transaction event.
          switch (t.getType()) {
              case PUT:
                  {
                      // Process the put operation here.
                  }
                  break;
              case DELETE:
                  {
                      // Process the delete operation here.
                  }
                  break;
              case TRANSACTION:
                  {
                      // Process the transaction operation here.
                  }
                  break;
              default:
                  // Received an unknown and therefore illegal operation
                  // type.
                  throw new
                  IllegalStateException("... exception message ...");
          }
      }
      }

  If row operations in the change stream include user-defined row metadata, this metadata is made available in the corresponding `StreamOperation` object received by `onNext()`. Subscribers can access this user-defined information attached to each row operation alongside the core data. This applies to all row changes performed through different APIs. For more information on user-defined row metadata, see Using user-defined row metadata in Developers Guide.

- `onComplete()`

  Signals the completion of a subscription. Use this method to perform whatever cleanup your application requires once a subscription has ended.

      @Override
      public void onComplete() {
      	/* nothing to do, so make this a no-op */
      }

  Note:

  You must implement this method in your stream processing application, because streaming from a KVStore table is unbounded by nature, so `onComplete()` will never be called. Any `no-op` implementation of this method will be ignored.

- `onError()`

  Signals that the subscription encountered an irrecoverable error and has to be terminated. The argument passed to this method is a `java.lang.Throwable` class instance. Use this method to perform whatever actions you want to take in response to the error.

      @Override
      public void onError(Throwable t) {
      	logger.severe("Error: " + t.getMessage());
      }

- `onWarn()`

  Signals that the subscription encountered a potential issue in the service. The argument passed to this method is a `java.lang.Throwable` class instance. Use this method to perform whatever actions you want to take in response to the warning.

      @Override
      public void onWarn(Throwable t) {
      	logger.warning("Warning: " + t.getMessage());
      }

  A warning does not end the subscription. Warnings in the form of `ShardTimeoutException` are provided as a way to inform the application of non-disruptive service issues such as a particular shard not responding for an extended period of time.

- `onCheckPointComplete()`

  Signals when a previously requested checkpoint has been completed. Checkpoints are performed by calling `NoSQLSubscription.doCheckpoint()`. Note that if an error occurred, the subscription will lose the checkpoint but the subscription itself will not terminate, and will continue streaming. See Using Checkpoints.

  Call this method with two arguments:

  - `oracle.kv.pubsub.StreamPosition`

    Identifies the location in the stream where the checkpoint was performed.

  - `java.lang.Throwable`

    `Null`, unless an error occurred while taking the checkpoint.

      @Override
      public void onCheckpointComplete(StreamPosition pos,
                                       Throwable cause) {
      	if (cause == null) {
      		logger.info("Finish checkpoint at position " + pos);
      	} else {
      		logger.warning("Fail to checkpoint at position " + pos +
      					   ", cause: " + cause.getMessage());
      	}
      }

- `onChangeResult()`

  Adding and removing tables from running subscription streams are made using asynchronous calls. The asynchronous calls will return immediately without any return value. The result of the operation can be fetched using the `onChangeResult` callback method after the change is effective. If the change was successful, this method will be called with a non-null stream position that represents the first stream position for which the change has taken effect. If the change was unsuccessful, but the subscription is still active, this method will be called with a non-null exception that describes the cause of the failure. If the change caused the subscription to be canceled, this method will not be called, and the `onError` method will be called instead.

- `getSubscriptionConfig()`

  Use this method to return the `oracle.kv.pubsub.NoSQLSubscriptionConfig` object used by this subscription. This method is invoked by the publisher when it is creating a subscription.