Domino meets RXJava

Verse on premises (VoP) is nearing its second beta release and fellow Notes experts are wondering if they need to install Apache Solr as part of the VoP deployment. There was a lengthy, high quality discussion and quite some effort evaluating alternatives. In conclusion it was decided to deliver the subset of Solr capabilities needed for VoP as series of OSGi plugins to the Domino server. The project was formed out of the experience with ProjectCastle, which continues as Project OrangeBox to deliver these plugins. In VoP you might encounter one or the other reference to PoB, so now you know where it comes from.
One of the design challenges to solve was to emulate the facet results of the Solr search engine. I build some prototypes and finally settled on the use of RxJava.
RxJava is a member of the ReactiveX programming family, which is designed around the Observer pattern, iterators and functional programming. Check out the main site to get into the groove.
The task at hand is to convert something Domino (a ViewNavigator, a DocumentCollection or a Document) into something that emits subscribable events. I started with turning a document into an NotesItem emitter. Purpose of this was the creation of lighweight Java objects that contain the items I'm interested in. Since Domino's Java has special needs and I couldn't use the ODA, special precaution was needed.
There are plenty of methods to create an Observable and on first look Create looks most promising, but it left the question of recycling open. Luckily there is the Using method that creates a companion object that lives along the Observable and gets explicitly called when the Observable is done. To create the NotesItem emitting Observable I settled on the From method with an Iterable as source. The moving parts I had to create were class DocumentSource implements Iterable<Item> and class ItemIterator implements Iterator<Item>
Why Reactive? In a nutshell: a source emits data and any number of subscribers can subscribe to. Between the emission and subscription any number of filters, modifiers and aggregators can manipulate the data emitted. Since each of them lives in its own little class, testing and composition become very easy. Let's look at an example:
docSource.getItemStream(session).filter(nameFilter).map(toPobItem).map(nameMapper).subscribe(new ItemAdder());
You almost can read this aloud: " The source emits a stream of items, they get filtered by Name, then converted into another Java object (PobItem) and renamed before added to the subscriber.". In a different case you might want to collect all entities (users, groups, roles) that have access to a document, you migh create a "readerAuthorFilter". The individual classes are very easy to test. E.g. the Name filter looks like this:

// requiredFields is is a Collection<String> of NotesItem names to include or exclude
ItemNameFilter nameFilter = new ItemNameFilter(requiredFields, ItemNameFilter.FilterMode.INCLUDE);

public class ItemNameFilter implements Func1<Item, Boolean> {

    public enum FilterMode {
        INCLUDE, EXCLUDE;
    }

    private final Logger      logger      = Logger.getLogger(this.getClass().getName());
    private final Set<String> itemNameSet = new HashSet<String>();
    private final FilterMode  filterMode;

    /**
     * Flexible include or exclude
     *
     * @param itemNames
     *            Collection of Names to include or exclude
     * @param filterMode
     *            INCLUDE or EXCLUDE
     */
    public ItemNameFilter(Collection<String> itemNames, FilterMode filterMode) {
        this.filterMode = filterMode;
        this.updateItemNames(itemNames);
    }

    public ItemNameFilter(Collection<String> itemNames) {
        this.filterMode = FilterMode.INCLUDE;
        this.updateItemNames(itemNames);
    }

    private void updateItemNames(Collection<String> itemNames) {
        this.itemNameSet.addAll(itemNames);
    }

    @Override
    public Boolean call(Item incomingItem) {
        // Include unless proven otherwise
        boolean result = true;
        try {
            String itemName = incomingItem.getName();
            boolean inList = this.itemNameSet.contains(itemName.toLowerCase());
            result = (inList && this.filterMode.equals(FilterMode.INCLUDE));
        } catch (NotesException e) {
            this.logger.log(Level.SEVERE, e);
            result = false;
        }
        return result;
    }
}

##
The two classes are not very long and, once you are comfortable with Reactive, easy to read:

ItemIterator

public class ItemIterator implements Iterator<Item> {
    @SuppressWarnings("rawtypes")
    private final Vector theNotesItems;
    private int          itemCount = 0;
    private Item         currentItem = null;

    @SuppressWarnings("rawtypes")
    public ItemIterator(Vector vector) {
        this.theNotesItems = vector;
    }

    @Override
    public boolean hasNext() {
        return ((!this.theNotesItems.isEmpty()) && (this.itemCount < this.theNotesItems.size()));
    }

    @Override
    public Item next() {
        // Recycle the current item. Shred takes care of null values
        Utils.shred(this.currentItem);
        his.currentItem = (Item) this.theNotesItems.get(this.itemCount);
        this.itemCount++;
        return result;
    }

    @Override
    public void remove() {
        throw new UnsupportedOperationException();
    }

}

DocumentSource

public class DocumentSource implements Iterable<Item> {
    private final Logger logger = Logger.getLogger(this.getClass().getName());
    private Document     doc    = null;
    private final String notesURL;

    public DocumentSource(final Session session, final Document doc) throws SourceNotAvailableException {
        this.doc = doc;
        try {
            this.notesURL = doc.getNotesURL();
        } catch (final NotesException e) {
            Utils.logError(this.logger, e);
            throw new SourceNotAvailableException("Can't retrieve NotesURL");
        }
    }

    /**
     * RxJava Observable access to all Items in a Notes Document
     *
     * @param session
     *            A Domino session to access the document if needed
     * @return a Observable with NotesItems
     */
    public Observable<Item> getItemStream(final Session session) {

        final Observable<Item> itemStream = Observable.using(new Func0<DocumentSource>() {
            @Override
            public DocumentSource call() {
                return DocumentSource.this;
            }
        }, new Func1<DocumentSource, Observable<Item>>() {
            @Override
            public Observable<Item> call(final DocumentSource documentSource) {
                return Observable.from(documentSource);
            }
        },

        new Action1<DocumentSource>() {
            @Override
            public void call(final DocumentSource documentSource) {
                documentSource.tearDown();
            }
        });

        return itemStream;
    }

    public String getNotesURL() {
        return this.notesURL;
    }

    @Override
    @SuppressWarnings("rawtypes")
    public Iterator<Item> iterator() {
        Vector result;
        try {
            result = this.doc.getItems();
        } catch (final NotesException e) {
            Utils.logError(this.logger, e);
            // Using an empty vector instead
            result = new Vector();
        }
        return new ItemIterator(result);
    }

    /**
     * Cleaning up a document source to free up the memory
     * allocated in the C API
     */
    protected void tearDown() {
        Utils.shred(this.doc);
        this.doc = null;
    }

}

I will report more on my RXJava meets Domino experience. As usual YMMV