src/main/java/me/paulbares/subscription/ApproverWorkflowRegistrar.java
public Subscription subscribe(CamundaUserDetails userDetails, Subscriber<Notification> subscriber) {
    String userId = userDetails.getUser();
    Collection<String> groupIds = userDetails.getGroups();
    writeExecute(userId, () -> {

      /*
       * Add the subscriber first before fetching the existing notifications to make sure if a new notification
       * arrives in parallel the subscriber does not miss the new notification.
       *
       *    Thread 1                                               | Thread 2
       * -  subscribe and add subscriber                           | new notification is created, onCreate is called but
       *                                                             the notification is not saved yet
       * -  getUnreadAndActiveNotificationsInDescOrder() is called | ø
       *    the new notification is not retrieved                  |
       * -  ø                                                      | new notification saved, it will be published to
       *                                                           | subscribers
       * -  ø                                                      | collect the subscribers to notify. The new
       *                                                           | subscriber is in the list.
       * -  send fetched notifications                             | send new notification
       *
       * The subscriber won't miss any notification, he might receive them in the wrong order but can reorder them
       * thanks to the notification timestamp.
       */

      this.subscribersByUserId.addSubscriber(userId, subscriber);
      groupIds
              .forEach(group -> this.subscribersByGroupId.computeIfAbsent(group, __ -> new Entry())
                      .addSubscriber(userId, subscriber));

      // Once registered, send initial view to subscriber
      List<Notification> notifications = this.notificationProvider.apply(userId, groupIds);
      Optional<Notification> max = notifications.stream().max(Comparator.comparingLong(Notification::getId));
      Long previous = this.lastSubIdBySub.put(subscriber, max.isPresent() ? max.get().getId() : -1);
      if (previous != null) {
        throw new IllegalStateException("A subscriber cannot be use multiple times");
      }

      if (!notifications.isEmpty()) {
        subscriber.onSubscribe(notifications);
      }
    });
    return new Subscription(userDetails, subscriber, this::unsubscribe);
  }
/**
   * Registers a new {@link Subscriber} for the given userDetails. It will receive all notifications intended to this
   * user and the groups he belongs to. After the subscription, the subscriber will receive all notifications.
   *
   * @param userDetails the user details
   * @param subscriber the subscriber to register
   * @return the {@link Subscription} to be used to stop receiving notifications by {@link Subscription#unsubscribe()
   *         unsubscribing}
   */
Registers a new  Subscriber for the given userDetails. It will receive all notifications intended to this
user and the groups he belongs to. After the subscription, the subscriber will receive all notifications.

@param userDetails the user details
@param subscriber the subscriber to register
@return the  Subscription to be used to stop receiving notifications by {@link Subscription#unsubscribe()
unsubscribing}
Registers a new Subscriber for the given userDetails . It will receive all notifications intended to this user and the groups he belongs to . After the subscription the subscriber will receive all notifications . @param userDetails the user details @param subscriber the subscriber to register @return the Subscription to be used to stop receiving notifications by { @link Subscription#unsubscribe () unsubscribing }
public Subscription subscribe(CamundaUserDetails userDetails, Subscriber<Notification> subscriber) {
    String userId = userDetails.getUser();
    Collection<String> groupIds = userDetails.getGroups();
    writeExecute(userId, () -> {
      /*
       * Add the subscriber first before fetching the existing notifications to make sure if a new notification
       * arrives in parallel the subscriber does not miss the new notification.
       *
       *    Thread 1                                               | Thread 2
       * -  subscribe and add subscriber                           | new notification is created, onCreate is called but
       *                                                             the notification is not saved yet
       * -  getUnreadAndActiveNotificationsInDescOrder() is called | ø
       *    the new notification is not retrieved                  |
       * -  ø                                                      | new notification saved, it will be published to
       *                                                           | subscribers
       * -  ø                                                      | collect the subscribers to notify. The new
       *                                                           | subscriber is in the list.
       * -  send fetched notifications                             | send new notification
       *
       * The subscriber won't miss any notification, he might receive them in the wrong order but can reorder them
       * thanks to the notification timestamp.
       */
      this.subscribersByUserId.addSubscriber(userId, subscriber);
      groupIds
              .forEach(group -> this.subscribersByGroupId.computeIfAbsent(group, __ -> new Entry())
                      .addSubscriber(userId, subscriber));
      List<Notification> notifications = this.notificationProvider.apply(userId, groupIds);
      Optional<Notification> max = notifications.stream().max(Comparator.comparingLong(Notification::getId));
      Long previous = this.lastSubIdBySub.put(subscriber, max.isPresent() ? max.get().getId() : -1);
      if (previous != null) {
        throw new IllegalStateException("A subscriber cannot be use multiple times");
      }
      if (!notifications.isEmpty()) {
        subscriber.onSubscribe(notifications);
      }
    });
    return new Subscription(userDetails, subscriber, this::unsubscribe);
  }
public Subscription subscribe ( CamundaUserDetails userDetails , Subscriber < Notification > subscriber ) { String userId = userDetails . getUser ( ) ; Collection < String > groupIds = userDetails . getGroups ( ) ; writeExecute ( userId , ( ) -> { /*
       * Add the subscriber first before fetching the existing notifications to make sure if a new notification
       * arrives in parallel the subscriber does not miss the new notification.
       *
       *    Thread 1                                               | Thread 2
       * -  subscribe and add subscriber                           | new notification is created, onCreate is called but
       *                                                             the notification is not saved yet
       * -  getUnreadAndActiveNotificationsInDescOrder() is called | ø
       *    the new notification is not retrieved                  |
       * -  ø                                                      | new notification saved, it will be published to
       *                                                           | subscribers
       * -  ø                                                      | collect the subscribers to notify. The new
       *                                                           | subscriber is in the list.
       * -  send fetched notifications                             | send new notification
       *
       * The subscriber won't miss any notification, he might receive them in the wrong order but can reorder them
       * thanks to the notification timestamp.
       */ this . subscribersByUserId . addSubscriber ( userId , subscriber ) ; groupIds . forEach ( group -> this . subscribersByGroupId . computeIfAbsent ( group , __ -> new Entry ( ) ) . addSubscriber ( userId , subscriber ) ) ; List < Notification > notifications = this . notificationProvider . apply ( userId , groupIds ) ; Optional < Notification > max = notifications . stream ( ) . max ( Comparator . comparingLong ( Notification :: getId ) ) ; Long previous = this . lastSubIdBySub . put ( subscriber , max . isPresent ( ) ? max . get ( ) . getId ( ) : - 1 ) ; if ( previous != null ) { throw new IllegalStateException ( "A subscriber cannot be use multiple times" ) ; } if ( ! notifications . isEmpty ( ) ) { subscriber . onSubscribe ( notifications ) ; } } ) ; return new Subscription ( userDetails , subscriber , this :: unsubscribe ) ; }
Registers a new {@link Subscriber} for the given userDetails.
Registers a new { @link Subscriber } for the given userDetails .
// Once registered, send initial view to subscriber