Revision as of 12:53, 15 February 2011

The gCube Document Library (gDL) is a client library for storing, updating, deleting and retrieving document description in a gCube infrastructure.

The gDL is a high-level component of the subsystem of gCube Information Services and it interacts with lower-level components of the subsystem to support document management processes within the infrastructure:

• the gCube Document Model (gDM) defines the basic notion of document and the gCube Model Library (gML) implements that notion into objects;
• the objects of the gML can be exchanged in the infrastructure as edge-labelled trees, and the Content Manager Library (CML) can model such trees as objects and dispatch them to the read and write operations of the Content Manager (CM) service;
• the CM implements these operations by translating trees to and from the content models of diverse repository back-ends.

The gDL builds on the gML and the CML to implement a local interface of CRUD operations that lift those of the CM to the domain of documents, efficiently and effectively.

Preliminaries

The core functionality of the gDL lies in its operations to read and write documents. The operations trigger interactions with remote services and the movement of potentially large volumes of data across the infrastructure. This may have a non-trivial and combined impact on the responsiveness of clients and the overall load of the infrastructure. The operations have been designed to minimise this impact. In particular:

• when reading, clients can qualify the documents that are relevant to their queries, and indeed what properties of relevant documents should be actually retrieved. These retrieval directives are captured in the gDL by the notion of document projections.

• when reading and writing, clients can move large numbers of documents across the infrastructure. The gDL streams this I/O movements so as to make efficient use of local and remote resources. It then defines a facilities with which clients can conveniently consume input streams, produce output streams, and more generally filter one stream into an other regardless of their origin. The facilities are collected into the stream DSL, an embedded domain-specific language for stream processing.

Understanding document projections and the stream DSL is key to reading and writing documents effectively. We discuss these preliminary concepts first, and then consider their use as input and outputs of the operations of the gDL.

Projections

A projection is a set of constraints over the properties of documents in the gDM. It can be used to match documents, i.e. identify documents whose properties satisfy the constraints of the projection.
Projections and matching are used in the read operations of the gDL:

• as a means to characterise relevant documents (projections as types);
• as a means to specify what parts of relevant documents should be retrieved (projections as retrieval directives).

The constraints of a projection take accordingly two forms:

• include constraints apply to properties that must be matched and retrieved;
• filter constraints apply to properties that must be matched but not retrieved.

note: in both cases, the constraints take the form of 'predicates' of the Content Manager Library] (CML). The projection itself converts into a complex predicate which is amenable for processing by the Content Manager service in the execution of retrieval operations. In this sense, projections are a key part of the document-oriented layer that the gDL defines over lower-level components of the service subsystem for content management.

As a simple example, a projection may define an include constraint over the name of metadata elements and a filter constraint over the time of their last update.
It may then be used to:

• characterise documents with metadata elements that match both constraints;
• retrieve of those documents only the name of matching metadata elements, excluding any other document property, including other inner elements and their properties.

All projections in the gDL have the Projection interface, which can be used in element-generic computations to access their constraints. To build projections, however, clients deal with one of the following implementation of the interface:

• DocumentProjection
• MetadataProjection
• AnnotationProjection
• PartProjection
• AlternativeProjection

A further implementation of the interface:

• PropertyProjection

allows clients to express constraints on the generic properties of any of the elements of the gDM.

Simple Projections

Clients create projections with the factory methods of the Projections companion class (a static import improves legibility and is recommended):

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;...
DocumentProjection dp = document();
 
MetadataProjection mp = metadata();
 
AnnotationProjection annp = annotation();
 
PartProjection pp = part();
 
AlternativeProjection altp = alteranative();

The projections above do not specify any include or filter constraints on the elements of the corresponding type. For example, dp matches all documents, regardless of their properties, inner elements, and properties of their inner elements. Similarly, mp matches all metadata elements of any document, regardless of their properties, and pp matches all the parts of any document, regardless of their properties. Thus the factory methods of the Projections class return empty projections.

Clients may add include constraints to a projection with the method with() declared by all projection classes. For document projections, for example:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
DocumentProjection dp = document().with(NAME);

With the above, the client adds the simplest form of constraint, an existence constraint that requires the target elements to have given properties, here the document to have name. Since this is an include constraint, the client is expressing an interest only in this property, regardless of the existence and values of other properties. Used as a parameter in the read operations of the gDL, this projection is translated into a directive to retrieve only the names of document(s) that have one.

note: properties are conveniently represented by constants in the Projections class. The constants are not strings, however, but dedicated Property objects that are specific to the type of projection. Trying to use properties that are undefined for the type of elements targeted by the projection is illegal and the error is detected statically.

Existence constraints may be expressed at once on multiple properties, e.g.:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
DocumentProjection dp = document().with(NAME,LANGUAGE,BYTESTREAM);

Besides inclusion constraints, clients may specify filter constraints with the method where() on projections, e.g:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
DocumentProjection dp = document().where(NAME,LANGUAGE);

Now, the client still requires documents to have a name and a language but he retains an interest in the other properties of matching documents. Used as a parameter in the read operations of the gDL, this projection is translated into a directive to retrieve all the properties of documents with a name.

Include and filter constraints can be combined, and the projections classes follow a builder pattern to add readability to the combinations. In particular, with() and where() return the very projection on which they are invoked. They may then be used as follows:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
DocumentProjection dp = document().with(NAME,SCHEMA_URI)
                                  .where(BYTESTREAM);

Here, the client requires documents to have a name and embed a bytestream that conforms to a schema, but he has an interest in processing only document names and schema URIs (e.g. for display purposes). Used as a parameter in the read operations of the gDL, this projection retrieves the requested information but avoids the transmission of bytestreams.

Optional Modifiers

Moving now beyond the simple existence of properties, another common requirement is to indicate the optionality of properties. Clients may wish to include certain properties, or equivalently filter by certain properties, if and only if these actually exists. In this case, clients can use the opt() of the Projections class as a constraint modifier, as this example illustrates:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
DocumentProjection dp = document().with(NAME,opt(SCHEMA_URI))
                                  .where(BYTESTREAM);

This projection differs from the previous one only because of the optionality constraint on the existence of a schema for the document's bytestream. Used as a parameter in the read operations of the gDL, this projection retrieves the name all documents that include a bytestream, but also their schema URI if they happen to have one.

A common use of optional modifier is with bytestream, which clients may wish either to find included in the document or else referred to with a URL:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
DocumentProjection dp = document().with(opt(BYTESTREAM),opt(URL));

Used as a parameter in the read operations of the gDL, this projection retrieves at most the bytestream and its URL for those documents that have both, only one of the two if the other is missing, and nothing at all if they are both missing.

note: The API allows optional modifiers in filter constraints too, but their application is rather pointless in this context (they will never elements from retrieval).

Deep Projections

In the examples above, we have considered existence constraints on simple element properties. The examples generalise easily to repeated structured properties, such as generic properties for all elements and inner element properties for documents.

Consider the following example:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
DocumentProjection dp = document().with(PART, opt(METADATA), PROPERTY);

Here the client adds three include constraints to the projection, all three for the existence of repeated properties. Documents that match this projection have at least one part, at least one generic property, and zero or more metadata elements. Used as a parameter in the read operations of the gDL, this projection retrieves all' the parts and all the generic properties of documents that have at least one of each, as well as all of their the metadata elements if they happen to have some.

Repeated properties such as generic properties and inner elements are also structured, i.e. have properties of their own. Clients that wish to constrain those properties too can use deep projections, i.e. embed within the projection of a given type one or more projections built for the structured properties of elements of that type. The following example illustrates the concept for metadata elements:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
MetadataProjection mp = meatadata().with(LANGUAGE).where(BYTESTREAM);
 
DocumentProjection dp = document().with(NAME, PART)
                                  .with(METADATA,mp);

The first projection constraints the existence of language and bytestream for metadata elements. The second projection constraints the existence of name and parts for document, as well as the existence of metadata elements that match the constraints of the first projection. The usual implications of include constraints and filter constraints apply. Used as a parameter in the read operations of the gDL, this projection retrieves the name, parts, and metadata elements of documents that have a name, at least one part, and at least one metadata element that includes a bystream. For the metadata elements, in particular, it retrieves only the language property.

Note that optionality constraints apply to deep projections as well as they apply to flat projections, as the following example shows:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
MetadataProjection mp = meatadata().with(LANGUAGE).where(BYTESTREAM);
 
DocumentProjection dp = document().with(NAME, PART)
                                  .with(opt(METADATA,mp));

This projection differs from the previous one only because the existence of on metadata elements that match the inner projection is optional. Documents that have a name and at least one part match the outer projection even if the have no metadata elements that match the inner projection (or no metadata elements at all).

Projections over Generic Properties

Generic properties are repeated and structured properties common to all elements. As for other properties with these characteristics, clients may wish to build deep projections that constraints their inner properties. For this purpose, the class Projections includes a dedicated factory method property(), as well as as specialised methods to express constraints. The following example illustrates the approach:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
...
 
PropertyProjection pp = property().withKey("somekey").with(PROPERTY_TYPE);
 
DocumentProjection dp = document().with(NAME, PART)
                                  .with(PROPERTY,pp);

Here, the client creates a document projection and embeds in it an inner projection that constrains its generic properties. The inner projection uses the method with() to add an include constraint for the existence of a type for the generic property, as usual. It also adds an include constraint to specify an exact value for the key of a generic property of interest. This relies on a method withKey() which is specific to projection over generic properties of elements. The reason for this specific construct is that, differently from other constrainable properties of elements, they key of a generic property serves as its identifier.

For the rest, property projections behave like other projections (e.g. can be used with optional modifiers). Used as a parameter in the read operations of the gDL, the projection above matches documents with a name, at least one part, and a property with key somekey and some type.

Advanced Projections

In more advanced forms of projections, clients may wish to specify constraints on properties other than mere existence. In these cases, they can use overloads of with() and where() that take as parameters Predicates that capture the desired constraints. As mentioned above, predicates are defined in the CML and gDL clients need to become acquainted with the range of available predicates and how to build them.

note: Deep projections already make use of this customisability. When clients embed a projection into another, they constrain the corresponding structured property with the predicate into which the inner projection translates.

Commonly, clients may wish to constrain the value of a property, as in the following example:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
import static org.gcube.contentmanagement.contentmanager.stubs.model.constraints.Constraints.*;import static org.gcube.contentmanagement.contentmanager.stubs.model.predicates.Predicates.*;...
DocumentProjection p = document().with(LANGUAGE,text(is("it"));

The client uses here the predicate text(is("it")) to constrain the language of documents to match the ISO639 code for the Italian language. As documented in the CML, the client builds the predicate with the static methods of the Predicates and Constraints classes, which he previously imports.

note: in building predicate expressions with the API of the CML, clients take responsibility for associating properties with predicates that are compatible with their type. In the example above, the language of an element is a textual property and thus only text()-based predicates can successfully match it. The gDL relinquishes the ability to ensure the correct construction of projections so as to allow clients to use the full expressiveness of the predicate language of the CML.

The type of constraints that can be expressed on properties is thus bound by the expressiveness of the predicate language of the CML. We include here another example to illustrate some of the possibilities:

import static org.gcube.contentmanagement.gcubedocumentlibrary.projections.Projections.*;
import static org.gcube.contentmanagement.contentmanager.stubs.model.constraints.Constraints.*;
import static org.gcube.contentmanagement.contentmanager.stubs.model.predicates.Predicates.*;
...
Calendar from = ...
Calendar to = ....
DocumentProjection p = document().with(URL,uri(matches("^ftp.*")));
                                 .where(CREATION_TIME,date(all(after(from),before(to))));

This projection is matched by documents that have been created at some point in between two dates, and with a bytestream available at some ftp server. Used as a parameter in the read operations of the gDL, the projection would retrieve only the URL of (the bytestream of) matching documents.

Streams

In some of its operations, the gDL relies on streams to model, process, and transfer inputs and outputs of potentially large size. Streams may consist of document descriptions, document identifiers, document updates, and more generally the outcomes of operations that take in turn large-scale inputs. Streamed processing makes efficient use of both local and remote resources, from local memory to network bandwidth, promoting the overall responsiveness of clients and services through reduced latencies.

Clients that make use of these operations will need to route streams towards and across the operations of the gDL, converting across stream interfaces and application logic in the process. As a common example, a client may need to route a remote result set of document identifiers to the read operations of the gDL, process the descriptions of the returned documents so as to update some of their properties, then feed the modified document descriptions to the write operations of the gDL so as to update them within the system, and finally inspect the outcomes of the updates so as to report or otherwise handle the failures that may have occurred in the process.

Throughout the workflow, it is important that the client remains within the paradigm of streamed processing, avoiding the accumulation of data in memory in all cases but where strictly induced by processing requirements. Document identifiers will be streaming from the remote location of the original result set as documents descriptions will be flowing back from yet another remote location, updated document descriptions will be leaving towards the same remote location as failures will be steadily coming back for handling.

Stream processing raises significant opportunities for clients, as well as non-trivial challenges. In recognition of the difficulties, the gDL includes a set of general-purpose facilities for stream processing that simplify the tasks of converting, filtering, transforming, or otherwise processing streams. These facilities are available as an embedded, domain-specific language, the Stream DSL.

Standard and Remote Iterators

As all the sentences of the Stream DSL take and return streams, we begin by looking look at how streams are represented in the operations of the gDL and the Stream DSL.

Streams have the interface of iterators, i.e. yield elements on demand and typically consumed within loops. There are two such interfaces:

• Iterator<T>, the standard Java interface for iterations.
• RemoteIterator<T>, a variation over Iterator<T> which makes explicit the remote origin of the stream.

In particular, a RemoteIterator differs from a standard Iterator in two respects:

• • the method next() may throw a checked Exception. This witnesses to the fact that iterating over the stream involves fallible I/O operations;
  • there is a method locator() that returns a reference to the remote stream as a plain String with implementation-specific syntax.

Locators aside, the key difference between the two interfaces is in their assumptions about the possibility of failures during iteration. A standard Iterator does not present failures to its clients other than for requests made past end of the stream (an unchecked NoSuchElementException). This may be because failures do not occur at all, e.g. the iteration is over an in-memory collection; it may also be because failures can occur but the iterator knows how to handle them. In this sense, Iterator<T> may well be defined over external, even remote collections, but it assumes that all failure handling policies are responsibilities of its implementations. In contrast, RemoteIterator<T> makes it clear that failures are likely to occur and that clients are expected to deal with them.

The operations of the gDL make use of both interfaces:

• when they take streams in input they expect them a standard Iterators;
• when they return streams in output the provide them as RemoteIterators.

This choice emphasises two points:

• streams that are provided by clients are of unknown origin, those provided by the library originate in remote services of the gCube Content Management infrastructure.
• all fault handling policies are in the hands of clients, where they should be. When they provide an Iterator to the library, they will have embedded a fault handling policy in its implementation. When they receive a RemoteIterator from the library, they will apply a fault handling policy at the point of stream consumption.

Stream Conversions and Fault Handling

The sentences of the stream DSL begin with verbs, which can be statically imported from the Streams class:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...

The verb convert introduces the simplest of sentences, those that convert between Iterators and RemoteIterators. The following example shows the conversion of an Iterator into a RemoteIterator:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
Iterator<SomeType> it = ...
RemoteIterator<SomeType> rit = convert(it);

The result is a RemoteIterator that promises to return failures but never does. The implementation is just a wrapper around the standard Iterator which returns it.toString()<code> as the locator of the underlying collection.

Converting a code>RemoteIterator to an Iterator is more interesting because it requires the encapsulation of a fault handling policy. The following example shows the possibilities:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<SomeType> rit = ...
 
//iterator will return any fault raised by the remote iterator
Iterator<SomeType> it1 = convert(rit).with(IGNORE_POLICY); 
//iterator will stop at the first fault raised by the remote iterator
Iterator<SomeType> it2 = convert(rit).with(FAILFAST_POLICY); 
//iterator will handle fault as specified by given policy
FaultPolicy policy = new FaultPolicy() {...}; 
Iterator<SomeType> it3 = convert(rit).with(policy);

In this example, the clause with() introduces the fault handling policy to encapsulate in the resulting Iterator. Two common policies are predefined and can be named directly, as shown for it1 and it2 above:

• IGNORE_POLICY: any faults raised by the RemoteIterator are discarded by the resulting Iterator<code>, which will ensure that <code>hasNext()>/code> and <code>next() behave as if they had not occurred;
• FAILFAST_POLICY: the first fault raised by the RemoteIterator halts the resulting Iterator, which will ensure that hasNext()>/code> and <code>next() behave as if they stream had reached its natural end;

Custom policies can be defined by implementing the interface FaultPolicy:

public interface FaultPolicy ... {
 
	boolean onFault(Exception e, int count); 
}

In onFault(), clients are passed the fault raised by the RemoteIterator, as well as the count of faults raised so far during the iteration (this will be greater than 1 only if the policy will have tolerated some previous faults during the iteration). Clients apply the policy and return true if the fault should be tolerated and the iteration continue, false if they instead wish the iteration to stop. Here's an example of a fault handling policy that tolerates only the first error and uses two aliases for the boolean values to improve the legibility of the policy (CONTINUE and STOP, also defined in the Streams class and statically imported):

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
FaultPolicy policy = new FaultPolicy() {
 
       public boolean onFault(Exception e, int count) {
             if (count=1) {
                   ....dealing with fault ...
		   return CONTINUE;
	      }
             else 
                  return STOP;	
        }
};

Note also that the IGNORE_POLICY is the default policy from conversion to standard iterators. Clients can use the clause withDefaults() to avoid naming it.

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<SomeType> rit = ...
 
//iterator will handle faults with the default policy: IGNORE_POLICY
Iterator<SomeType> it = convert(rit).withDefaults();

Finally, note that stream conversions may also be applied between RemoteIterators, so as to change their FaultPolicy:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<SomeType> rit1 = ...
 
//iterator will handle faults with the default policy: IGNORE_POLICY
RemoteIterator<SomeType> rit2 = convert(rit1).withRemote(IGNORE_POLICY);

Here, the clause withRemote() introduces a fault policy for the RemoteIterator in output. Fault policies for RemoteIterator are a superset of those that can be configured on standard Iterators. In particular, they implement the interface RemoteFaultPolicy:

public interface RemoteFaultPolicy ... { 
	boolean onFault(Exception e, int count) throws Exception; 
}

Note that the only difference between a FaultPolicy and a RemoteFaultPolicy is that the latter has the additional option to raise a fault of its own in onFault(). Thus, when a fault occurs during iteration, the RemoteIterator can continue iterating, stop the iteration, but also re-throw the same or another fault to the iterating client, which is indeed what makes a RemoteIterator different from a standard Iterator.

In particular, the Stream DSL predefines a third policy which is available only for RemoteIterators:

• RETHROW_POLICY: any faults raised during iteration will be immediately propagated to clients;

This is the in fact the default policy for RemoteIterators and clients can use the clause withRemoteDefaults() to avoid naming it:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<SomeType> rit1 = ...
 
RemoteIterator<SomeType> rit2 = convert(rit1).withRemoteDefaults();

ResultSet Conversions

A different but very common form of conversion is between gCube result sets and RemoteIterators. While result sets are the preferred way of modelling remote streams within the system, their iterators do not natively implement the RemoteIterator<T> interface, which has been independently defined in the CML as an abstraction over an underlying result set mechanism. The CML defines an initial set of facilities to perform the conversion from result sets of untyped string payloads to RemoteIterators of typed objects. The Stream DSL builds on these facilities to cater for a few common conversions:

• documentsIn(RSLocator): takes the locator to a result set of serialisations of GCubeDocuments and returns a RemoteIterator<GCubeDocument>;
• metadataIn(RSLocator): takes the locator to a result set of serialisations of GCubeDocuments and returns a RemoteIterator<GCubeMetadata>;
• annotationsIn(RSLocator): takes the locator to a result set of serialisations of GCubeDocuments and returns a RemoteIterator<GCubeAnnotation>;
• partsIn(RSLocator): takes the locator to a result set of serialisations of GCubeDocuments and returns a RemoteIterator<GCubePart>;
• alternativesIn(RSLocator): takes the locator to a result set of serialisations of GCubeDocuments and returns a RemoteIterator<GCubeAlternative>;
• payloadsIn(RSLocator): takes the locator to a result set and returns a RemoteIterator<String>.

Essentially, documentsIn() adapts the result set to a RemoteIterator<T> that parses documents as it iterates over their serialisations. The following methods do the same, but extract the corresponding GCubeElements from the GCubeDocuments obtained from parsing. All the methods are based on the last one, payloadsIn, which is also immediately useful to feed result set of GCubeDocument identifiers to the read operations the gDL that perform stream-based document lookups.

note: all the conversions above produce RemoteIterators that return the locator of the original result set from invocations of locator(). Clients can use the locator to process the stream with standard set-based APIs, as usual.

The usage pattern is straightforward and combines with the previous conversions. The following example illustrates:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RSLocator rs = ...
Iterator<GCubeDocument> it = convert(documentsIn(rs)).with(FAILFAST_POLICY);

Pipes and Filters

The conversions introduced above do not alter the original streams, i.e. the output iterators produce the same elements of the input iterators. The exception is with result set-based conversions: documentsIn() parses the untyped payloads of the input result sets into typed objects, while methods such as metadataIn() extract GCubeMetadata elements from GCubeDocuments. Parsing and extraction are only examples of the kind of post-processing that clients may wish to apply to the elements of existing stream to produce a new stream of post-processed elements. All the remaining sentences of the Stream DSL are dedicated precisely to this kind of tasks.

Sentences introduced by the verb pipe take a stream and return a second stream that applies an arbitrary filter to the elements of the first stream, encapsulating a fault handing policy in the process. The following example illustrates basic usage:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
Iterator<GCubeDocument> it1 = ...
 
Filter<GCubeDocument,String> filter = new Filter<GCubeDocument,String>() { 
                  public String apply(GCubeDocument doc) throws Exception {                           return doc.name();
                  }
};
 
Iterator<GCubeDocument> it2 = pipe(it1).though(filter).withDefaults();

Here, a standard Iterator of GCubeDocuments is piped through a filter that extracts the names of GCubeDocuments. The result is another standard Iterator that produces document names from the original stream. The clause through() introduces the filter on the output stream and, as already discussed for conversion methods, the clause withDefaults() configures the default IGNORE_POLICY for it.

As shown in the example, filters are implementations of the Filter<FROM,TO> interface. The method apply() is self-explanatory: clients are given the elements of the unfiltered stream as the filtered stream is being iterated over, and they have the onus to produce and return an element of the filtered stream. The only point worth stressing is that apply()s can throw a fault if it cannot produce an element of the filtered stream. The filtered stream passes these faults to the FaultPolicy configured for it. In this example, faults clearly cannot occur. If they did, however, the configured policy would simply ignore them, i.e. the problematic elements of the input stream would not contribute to the contents of the filtered stream.

In the example the input stream and the filtered one are both standard Iterators. The construct, however, is generic and can be used to filter any form of stream into any other. In this sense, the construct embeds stream conversions within its clauses. As an example, consider the common case in which a RemoteIterator is filtered into a standard Iterator:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<GCubeDocument> rit = ...
 
Filter<GCubeDocument,SometType> filter = ....;
 
Iterator<SomeType> it = pipe(rit).though(filter).with(FAILFAST_POLICY);

Here, filter is applied to the elements of a RemoteIterator to produce a standard Iterator that stops as soon as the input stream raises a fault. Conversely, in the following example:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<GCubeDocument> rit1 = ...
 
Filter<GCubeDocument,SometType> filter = ....;
 
RemoteIterator<SomeType> rit2 = pipe(rit1).though(filter).withRemote(IGNORE_POLICY);

Here, filter is applied to the elements of a RemoteIterator to produce yet another RemoteIterator that ignores any fault raised by the input iterator.

To conclude with pipe-based sentences, note that the Stream DSL includes Processor<T>, a base implementation of Filter&ltFROM,TO> that simplifies the declaration of filters that simply mutate the input and return it. The following example illustrates usage:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<GCubeDocument> rit1 = ...
 
Processor<GCubeDocument> processor = new Processor<GCubeDocument>() { 
            public void process(GCubeDocument doc) throws Exception {                       doc.setName(doc.name()+"-modified");
} ;
 
RemoteIterator<GCUBEDocument> rit2 = pipe(rit1).though(processor).withRemoteDefaults();

Here, the processor simply updates the GCubeDocuments in the input stream by changing their name. The output stream thus returns the same elements of the input stream, albeit updated. During iteration, its policy is simply to re-throw any fault that may be raised by the input iterator.

Folding and Unfolding

With pipe-based sentences, clients can filter the elements of a stream into the elements of another streams. While the elements of the two stream can vary arbitrarily in type, the correspondence between elements of the two streams is fairly strict: for each element of the input stream there may be at most one element of the output stream (elements that raise iteration failures in the input stream may have no counterpart in the output stream, i.e. may be discarded). In this sense, the streams are always consumed in phase.

In some cases, however, clients may wish to:

• fold a stream, i.e. produce another stream that has one List element for each N elements of the original stream;
• unfold a stream, i.e. produce another stream that has N elements for each element in the original stream.

Conceptually, these requirements are still within the scope of filtering, but the fact that the consumption of the filtered stream is out of phase with respect to the unfiltered stream requires a rather different treatment. For this reason, the Stream DSL offers two dedicated classes of sentences:

• group-based sentences for stream folding;
• unfold-based sentences for stream unfolding.

To fold a stream, clients indicate how many elements of the stream should be grouped into elements of the folded stream, what filter should be applied to each of the elements of the stream and, as usual, what fault handling policy should be used for the folded stream. The following example illustrates usage in the common case in which a RemoteIterator is folded into a standard Iterator:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<GCubeDocument> rit = ...
 
Filter<GCubeDocument,SometType> filter = ....;
 
Iterator<List<SomeType>> it = group(rit).in(10).pipingThrough(filter).withDefaults();

The RemoteIterator is here folded in Lists of 10 elements, (or smaller, if the end of the input stream is reached before a List of The clause in() indicates the maximum size of the output Lists. Each of the GCubeDocuments in the original stream is then passed through filter, which produces one of the List elements for it. The clause pipingThrough allows the configuration of the filer. Finally, the default IGNORE_POLICY is set on the folded stream with the clause withDefaults(), meaning that any fault raised by the RemoteIterator or filter will be tolerated and the element that caused the failure will simply not contribute to the accumulation of the next 10 elements of the folded stream.

note: the example shows the folding of a RemoteIterator into a standard Iterator but, as for all the sentences of the DSL, all combinations of input and output streams are possible, with the usual implications on the fault handing policies that can be set on the folded stream and with the optional choice of Processors over Filters in cases where folding simply groups updated elements of the stream.

It is a common requirement to fold a stream without transforming or altering otherwise its elements. In this case, the clause pipingThrough can be omitted altogether from the sentence:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<GCubeDocument> rit = ...
 
Iterator<List<GCubeDocument>> it = group(rit).in(10).withDefaults();

Effectively, the stream is here being filtered with a pass-through filter that simply returns the elements of the unfolded streams. As we shall see, t his kind of folding is particularly useful to 'slice' a stream in small in-memory collections that can be used with the write operations of the gDL that work in bulk and by-value.

Unfolding a stream follows a similar pattern, as shown in the following example:

import static org.gcube.contentmanagement.gcubedocumentlibrary.streams.dsl.Streams.*;
...
RemoteIterator<GCubeDocument> rit = ...
 
Filter<GCubeDocument,List<SometType>> filter = ....;
 
Iterator<SomeType> it = unfold(rit).pipingThrough(filter).withDefaults();

This time we cannot dispense with using a Filter, which is necessary to map a single element of the stream into a List of elements that the unfolded stream, a standard Iterator in this example, will then yield one at the time at the client's demand. As usual, all combinations of standard Iterators, RemoteIterators, and fault handling policies are allowed. Using Processors is instead disallowed here, as it's in the nature of unfolding to convert a element into a number of different elements. Unfolding and updates, in other words, do not interact well.

The most common application of unfolding is for the extraction of inner elements from documents, e.g. unfold a stream of GCubeDocuments into a stream of GCubeMetadata elements, where each element in the unfolded stream belongs to some GCubeDocument in the document stream. Accordingly, the Stream DSL predefines a comprehensive number of these unfoldings. We have seen some of them already, where the document input stream was in the form of a result set (e.g. metadataIn(RSLocator)). Similar unfoldings are directly available on RemoteIterator<GCubeDocument>s:

• metadataIn(RemoteIterator<GCubeDocument>);
• annotationsIn(RemoteIterator<GCubeDocument>);
• partsIn(RemoteIterator<GCubeDocument>);
• alternativesIn(RemoteIterator<GCubeDocument>);

Summary

In summary, the Stream DSL allows clients to formulate the following sentences for stream manipulation:

• convert between standard Iterators and/or RemoteIterators:
  • convert(Iterator);
  • convert(RemoteIterator).with(FaultPolicy);
  • convert(RemoteIterator).withDefaults();
  • convert(RemoteIterator).withRemote(RemoteFaultPolicy);
  • convert(RemoteIterator).withRemoteDefaults();
• convert between result sets and RemoteIterator:
  • payloadsIn(RSLocator);
  • documentIn(RSLocator);
  • metadataIn(RSLocator);
  • annotationsIn(RSLocator);
  • partsIn(RSLocator);
  • alternativesIn(RSLocator);
• pipe streams through filters:
  • pipe(Iterator).through(Filter|Processor).with(FaultPolicy);
  • pipe(Iterator).through(Filter|Processor).withDefaults();
  • pipe(RemoteIterator).through(Filter|Processor).with(FaultPolicy);
  • pipe(RemoteIterator).through(Filter|Processor).with(FaultPolicy);
  • pipe(Iterator).through(Filter|Processor).withRemote(RemoteFaultPolicy);
  • pipe(Iterator).through(Filter|Processor).withRemoteDefaults();
• fold streams:
  • group(Iterator).in(N).pipingThrough(Filter|Processor).with(FaultPolicy);
  • group(Iterator).in(N).pipingThrough(Filter|Processor).withDefaults();
  • group(Iterator).in(N).pipingThrough(Filter|Processor).withRemote(RemoteFaultPolicy);
  • group(Iterator).in(N).pipingThrough(Filter|Processor).withRemoteDefaults();
  • group(Iterator).in(N).with(FaultPolicy);
  • group(Iterator).in(N).withDefaults();
  • group(Iterator).in(N).withRemote(RemoteFaultPolicy);
  • group(Iterator).in(N).withRemoteDefaults();
  • group(RemoteIterator).in(N).pipingThrough(Filter|Processor).with(FaultPolicy);
  • group(RemoteIterator).in(N).pipingThrough(Filter|Processor).withDefaults();
  • group(RemoteIterator).in(N).pipingThrough(Filter|Processor).withRemote(RemoteFaultPolicy);
  • group(RemoteIterator).in(N).pipingThrough(Filter|Processor).withRemoteDefaults();
  • group(RemoteIterator).in(N).withRemote(FaultPolicy);
  • group(RemoteIterator).in(N).withDefaults();
  • group(RemoteIterator).in(N).withRemote(RemoteFaultPolicy);
  • group(RemoteIterator).in(N).withRemoteDefaults();
• unfold streams:
  • unfold(Iterator).pipingThrough(Filter).with(FaultPolicy);
  • unfold(Iterator).pipingThrough(Filter).withDefaults();
  • unfold(Iterator).pipingThrough(Filter).withRemote(RemoteFaultPolicy);
  • unfold(Iterator).pipingThrough(Filter).withRemoteDefaults();
  • unfold(RemoteIterator).pipingThrough(Filter).with(FaultPolicy);
  • unfold(RemoteIterator).pipingThrough(Filter).withDefaults();
  • unfold(RemoteIterator).pipingThrough(Filter).withRemote(RemoteFaultPolicy);
  • unfold(RemoteIterator).pipingThrough(Filter).withRemoteDefaults();
  • metadataIn(Iterator<GCubeDocument>);
  • annotationsIn(Iterator<GCubeDocument>);
  • partsIn(Iterator<GCubeDocument>);
  • alternativesIn(Iterator<GCubeDocument>);
  • metadataIn(RemoteIterator<GCubeDocument>);
  • annotationsIn(RemoteIterator<GCubeDocument>);
  • partsIn(RemoteIterator<GCubeDocument>);
  • alternativesIn(RemoteIterator<GCubeDocument>);

Operations

Reading Documents

Adding Documents

Updating Documents

Deleting Documents

Views

Transient Views

Persistent Views

Creating Views

Discovering Views

Using Views

Advanced Topics

Caches

Buffers