You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by to...@apache.org on 2015/04/29 02:27:29 UTC
[14/14] incubator-usergrid git commit: WIP 2 squash
WIP 2 squash
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/d8c0f811
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/d8c0f811
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/d8c0f811
Branch: refs/heads/USERGRID-593
Commit: d8c0f811a33639c25904e19d6fd881838e89adb9
Parents: d27cd8f
Author: Todd Nine <tn...@apigee.com>
Authored: Tue Apr 28 18:27:12 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Tue Apr 28 18:27:12 2015 -0600
----------------------------------------------------------------------
.../corepersistence/CpRelationManager.java | 11 +-
.../corepersistence/pipeline/DataPipeline.java | 140 -----------
.../corepersistence/pipeline/Pipeline.java | 123 ++++++++++
.../pipeline/PipelineContext.java | 93 ++++++++
.../pipeline/PipelineModule.java | 9 +-
.../pipeline/PipelineResult.java | 57 +++++
.../pipeline/cursor/CursorSerializerUtil.java | 6 -
.../pipeline/cursor/NoCursorSerializer.java | 55 -----
.../pipeline/cursor/RequestCursor.java | 1 -
.../pipeline/cursor/ResponseCursor.java | 15 +-
.../pipeline/read/AbstractFilter.java | 114 ---------
.../read/AbstractPipelineOperation.java | 44 ++++
.../pipeline/read/AbstractResumableFilter.java | 62 +++++
.../pipeline/read/CandidateResultsFilter.java | 2 +-
.../pipeline/read/Collector.java | 7 +-
.../pipeline/read/CollectorFactory.java | 44 ++++
.../corepersistence/pipeline/read/Filter.java | 35 +--
.../pipeline/read/FilterFactory.java | 97 ++++++++
.../corepersistence/pipeline/read/IdFilter.java | 30 ---
.../pipeline/read/PipelineOperation.java | 38 +++
.../pipeline/read/ReadFilterFactory.java | 100 --------
.../pipeline/read/ReadFilterFactoryImpl.java | 234 +++++++++----------
.../pipeline/read/ReadPipelineBuilder.java | 24 +-
.../pipeline/read/ReadPipelineBuilderImpl.java | 148 +++++++-----
.../pipeline/read/ResultsPage.java | 42 ++++
.../AbstractElasticSearchFilter.java | 15 +-
.../CandidateResultsEntityResultsCollector.java | 216 +++++++++++++++++
.../CandidateResultsIdVerifyFilter.java | 193 +++++++++++++++
.../CollectionElasticSearchFilter.java | 4 +-
.../ConnectionElasticSearchFilter.java | 4 +-
.../EntityLoadCandidateResults.java | 230 ------------------
.../pipeline/read/entity/EntityIdFilter.java | 15 +-
.../read/entity/EntityLoadCollector.java | 84 ++-----
.../graph/AbstractReadGraphEdgeByIdFilter.java | 17 +-
.../read/graph/AbstractReadGraphFilter.java | 8 +-
.../graph/ReadGraphConnectionByTypeFilter.java | 8 +-
.../results/ObservableQueryExecutor.java | 62 ++++-
.../corepersistence/StaleIndexCleanupTest.java | 2 +-
.../index/AsyncIndexServiceTest.java | 2 +-
.../corepersistence/index/IndexServiceTest.java | 2 +-
.../pipeline/cursor/CursorTest.java | 4 +-
.../usergrid/persistence/CollectionIT.java | 12 +-
.../org/apache/usergrid/persistence/GeoIT.java | 8 +-
.../apache/usergrid/persistence/IndexIT.java | 8 +-
.../PerformanceEntityRebuildIndexTest.java | 2 +-
.../query/IntersectionTransitivePagingIT.java | 2 +-
.../query/IntersectionUnionPagingIT.java | 2 +-
.../persistence/query/IteratingQueryIT.java | 30 +--
.../persistence/query/NotSubPropertyIT.java | 2 +-
.../persistence/index/impl/EntityIndexTest.java | 74 +++---
.../persistence/index/impl/GeoPagingTest.java | 2 +-
.../index/impl/IndexLoadTestsIT.java | 2 +-
52 files changed, 1430 insertions(+), 1111 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
index aa03c24..8c3865e 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CpRelationManager.java
@@ -32,7 +32,9 @@ import org.springframework.util.Assert;
import org.apache.usergrid.corepersistence.asyncevents.AsyncEventService;
import org.apache.usergrid.corepersistence.pipeline.PipelineBuilderFactory;
+import org.apache.usergrid.corepersistence.pipeline.PipelineResult;
import org.apache.usergrid.corepersistence.pipeline.read.ReadPipelineBuilder;
+import org.apache.usergrid.corepersistence.pipeline.read.ResultsPage;
import org.apache.usergrid.corepersistence.results.ObservableQueryExecutor;
import org.apache.usergrid.corepersistence.util.CpEntityMapUtils;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
@@ -642,7 +644,7 @@ public class CpRelationManager implements RelationManager {
}
- final Observable<Results> resultsObservable = readPipelineBuilder.build();
+ final Observable<PipelineResult<ResultsPage>> resultsObservable = readPipelineBuilder.execute();
return new ObservableQueryExecutor( resultsObservable ).next();
}
@@ -905,15 +907,12 @@ public class CpRelationManager implements RelationManager {
if ( query.isGraphSearch() ) {
readPipelineBuilder.getConnection( connection );
}
- else if ( entityType != null ) {
- readPipelineBuilder.connectionWithQuery( connection, query.getQl().get(), entityType );
- }
else {
- readPipelineBuilder.connectionWithQuery( connection, query.getQl().get() );
+ readPipelineBuilder.connectionWithQuery( connection, Optional.fromNullable( entityType ), query.getQl().get() );
}
- final Observable<Results> resultsObservable = readPipelineBuilder.build();
+ final Observable<PipelineResult<ResultsPage>> resultsObservable = readPipelineBuilder.execute();
return new ObservableQueryExecutor( resultsObservable ).next();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/DataPipeline.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/DataPipeline.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/DataPipeline.java
deleted file mode 100644
index f7abcfa..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/DataPipeline.java
+++ /dev/null
@@ -1,140 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.pipeline;
-
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.usergrid.corepersistence.pipeline.cursor.RequestCursor;
-import org.apache.usergrid.corepersistence.pipeline.cursor.ResponseCursor;
-import org.apache.usergrid.corepersistence.pipeline.read.Collector;
-import org.apache.usergrid.corepersistence.pipeline.read.Filter;
-import org.apache.usergrid.corepersistence.pipeline.read.IdFilter;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import com.google.common.base.Optional;
-
-import rx.Observable;
-
-
-/**
- * A pipeline that will allow us to build a traversal command for execution
- *
- * See http://martinfowler.com/articles/collection-pipeline/ for some examples
- *
- * TODO: Re work the cursor and limit phases. They need to be lazily evaluated, not added on build time
- */
-public class DataPipeline<R> {
-
-
- private final ApplicationScope applicationScope;
- private final List<Filter> idFilterList;
-
- private Optional<String> cursor;
- private int limit;
-
-
- private int count = 0;
-
-
- /**
- * Our first pass, where we implement our start point as an Id until we can use this to perform our entire
- * traversal. Eventually as we untangle the existing Query service nightmare, the sourceId will be remove and
- * should only be traversed from the root application
- */
- public DataPipeline( final ApplicationScope applicationScope, final List<Filter> filters, final Collector<?, R> collector ) {
-
- this.applicationScope = applicationScope;
-
-
- idFilterList = new ArrayList<>();
- }
-
-
- /**
- * Add a read command that will read Ids and produce Ids. This is an intermediate traversal operations
- */
- public DataPipeline withTraverseCommand( final IdFilter traverseCommand ) {
-
- this.idFilterList.add( traverseCommand );
-
- return this;
- }
-
-
- public Observable<R> execute(){
- return null;
- }
-
- /**
- * Build the final collection step, and process our filters
- */
- public <T, R> Observable<R> build( final Collector<T, R> pipeCollector ) {
-
- RequestCursor requestCursor = new RequestCursor( this.cursor );
- ResponseCursor responseCursor = new ResponseCursor();
-
- Observable<Id> traverseObservable = Observable.just( applicationScope.getApplication() );
-
- //build our traversal commands
- for ( Filter filter : idFilterList ) {
- setState( filter, requestCursor, responseCursor );
-
- traverseObservable = traverseObservable.compose( filter );
- }
-
-
- setState( pipeCollector, requestCursor, responseCursor );
-
- pipeCollector.setLimit( limit );
-
- return traverseObservable.compose( pipeCollector );
- }
-
-
- public void setCursor( Optional<String> cursor ) {
- this.cursor = cursor;
- }
-
-
- public void setLimit( final int limit ) {
- this.limit = limit;
- }
-
-
- /**
- * Set the id of the state
- */
- private void setState( final Filter filter, final RequestCursor requestCursor,
- final ResponseCursor responseCursor ) {
-
- //TODO, see if we can wrap this observable in our ObservableTimer so we can see how long each filter takes
-
-
- filter.setId( count );
- //done for clarity
- count++;
-
- filter.setCursorCaches( requestCursor, responseCursor );
- filter.setApplicationScope( applicationScope );
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/Pipeline.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/Pipeline.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/Pipeline.java
new file mode 100644
index 0000000..bc93b6c
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/Pipeline.java
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.pipeline;
+
+
+import java.util.List;
+
+import org.apache.usergrid.corepersistence.pipeline.cursor.RequestCursor;
+import org.apache.usergrid.corepersistence.pipeline.cursor.ResponseCursor;
+import org.apache.usergrid.corepersistence.pipeline.read.Collector;
+import org.apache.usergrid.corepersistence.pipeline.read.PipelineOperation;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+
+import com.google.common.base.Optional;
+
+import rx.Observable;
+
+
+/**
+ * A pipeline that will allow us to build a traversal command for execution
+ *
+ * See http://martinfowler.com/articles/collection-pipeline/ for some examples
+ *
+ * TODO: Re work the cursor and limit phases. They need to be lazily evaluated, not added on build time
+ */
+public class Pipeline<R> {
+
+
+ private final ApplicationScope applicationScope;
+ private final List<PipelineOperation> idPipelineOperationList;
+ private final Collector<?, R> collector;
+ private final RequestCursor requestCursor;
+ private final ResponseCursor responseCursor;
+
+ private final int limit;
+
+
+ private int idCount = 0;
+
+
+ /**
+ * Our first pass, where we implement our start point as an Id until we can use this to perform our entire
+ * traversal. Eventually as we untangle the existing Query service nightmare, the sourceId will be remove and
+ * should only be traversed from the root application
+ */
+ public Pipeline( final ApplicationScope applicationScope, final List<PipelineOperation> pipelineOperations,
+ final Collector<?, R> collector, final Optional<String> cursor, final int limit ) {
+
+ this.applicationScope = applicationScope;
+ this.idPipelineOperationList = pipelineOperations;
+ this.collector = collector;
+ this.limit = limit;
+
+ this.requestCursor = new RequestCursor( cursor );
+ this.responseCursor = new ResponseCursor();
+ }
+
+
+ /**
+ * Execute the pipline construction, returning an observable of results
+ * @return
+ */
+ public Observable<PipelineResult<R>> execute(){
+
+
+ Observable traverseObservable = Observable.just( applicationScope.getApplication() );
+
+ //build our traversal commands
+ for ( PipelineOperation pipelineOperation : idPipelineOperationList ) {
+ setState( pipelineOperation );
+
+ //TODO, see if we can wrap this observable in our ObservableTimer so we can see how long each filter takes
+
+
+ traverseObservable = traverseObservable.compose( pipelineOperation );
+ }
+
+
+ setState( collector );
+
+ final Observable<R> response = traverseObservable.compose( collector );
+
+
+ //append the optional cursor into the response for the caller to use
+ return response.map( result -> new PipelineResult<>( result, responseCursor ) );
+ }
+
+
+
+
+ /**
+ * Set the id of the state
+ */
+ private void setState( final PipelineOperation pipelineOperation ) {
+
+
+ final PipelineContext context = new PipelineContext( applicationScope, requestCursor, responseCursor,
+ limit, idCount );
+
+ pipelineOperation.setContext( context );
+
+ //done for clarity
+ idCount++;
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineContext.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineContext.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineContext.java
new file mode 100644
index 0000000..3a07fce
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineContext.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.pipeline;
+
+
+import java.io.Serializable;
+
+import org.apache.usergrid.corepersistence.pipeline.cursor.CursorSerializer;
+import org.apache.usergrid.corepersistence.pipeline.cursor.RequestCursor;
+import org.apache.usergrid.corepersistence.pipeline.cursor.ResponseCursor;
+import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+
+import com.google.common.base.Optional;
+
+
+/**
+ * Encapsulates the context of the pipeline for the scope of the filter.
+ */
+public class PipelineContext {
+
+ private final int id;
+ private final ApplicationScope applicationScope;
+ private final RequestCursor requestCursor;
+ private final ResponseCursor responseCursor;
+ private final int limit;
+
+
+ public PipelineContext( final ApplicationScope applicationScope, final RequestCursor requestCursor,
+ final ResponseCursor responseCursor, final int limit, final int id ) {
+
+ this.applicationScope = applicationScope;
+ this.requestCursor = requestCursor;
+ this.responseCursor = responseCursor;
+ this.limit = limit;
+ this.id = id;
+ }
+
+
+ public ApplicationScope getApplicationScope() {
+ return applicationScope;
+ }
+
+
+ public int getId() {
+ return id;
+ }
+
+
+ /**
+ * Get our cursor value if present
+ */
+ public <T extends Serializable> Optional<T> getCursor( final CursorSerializer<T> serializer ) {
+ final T value = requestCursor.getCursor( id, serializer );
+
+ return Optional.fromNullable( value );
+ }
+
+
+ /**
+ * Set the cursor value into our resposne
+ */
+ public <T extends Serializable> void setCursorValue( final T value, final CursorSerializer<T> serializer ) {
+ responseCursor.setCursor( id, value, serializer );
+ }
+
+
+ /**
+ * Get the limit for this execution
+ * @return
+ */
+ public int getLimit() {
+ return limit;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineModule.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineModule.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineModule.java
index 55b84af..3018718 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineModule.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineModule.java
@@ -20,7 +20,8 @@
package org.apache.usergrid.corepersistence.pipeline;
-import org.apache.usergrid.corepersistence.pipeline.read.ReadFilterFactory;
+import org.apache.usergrid.corepersistence.pipeline.read.CollectorFactory;
+import org.apache.usergrid.corepersistence.pipeline.read.FilterFactory;
import org.apache.usergrid.corepersistence.pipeline.read.ReadFilterFactoryImpl;
import org.apache.usergrid.corepersistence.pipeline.read.ReadPipelineBuilder;
import org.apache.usergrid.corepersistence.pipeline.read.ReadPipelineBuilderImpl;
@@ -38,7 +39,7 @@ public class PipelineModule extends AbstractModule {
protected void configure() {
//Use Guice to create the builder since we don't really need to do anything
//other than DI when creating the filters
- bind( ReadFilterFactory.class ).to( ReadFilterFactoryImpl.class );
+// bind( FilterFactory.class ).to( ReadFilterFactoryImpl.class );
//Use Guice to create the builder since we don't really need to do anything
@@ -47,9 +48,11 @@ public class PipelineModule extends AbstractModule {
.build( PipelineBuilderFactory.class ) );
+// install( new Factory)
//Use Guice to create the builder since we don't really need to do anything
//other than DI when creating the filters
-// install( new FactoryModuleBuilder().build( ReadFilterFactory.class ) );
+ install( new FactoryModuleBuilder().build( FilterFactory.class ) );
+ install( new FactoryModuleBuilder().build( CollectorFactory.class ));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineResult.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineResult.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineResult.java
new file mode 100644
index 0000000..fe8604e
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/PipelineResult.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.pipeline;
+
+
+import org.apache.usergrid.corepersistence.pipeline.cursor.ResponseCursor;
+
+import com.google.common.base.Optional;
+
+
+/**
+ * Intermediate observable that will return results, as well as an optional cursor
+ * @param <R>
+ */
+public class PipelineResult<R> {
+
+
+ private final R result;
+
+ private final ResponseCursor responseCursor;
+
+
+ public PipelineResult( final R result, final ResponseCursor responseCursor ) {
+ this.result = result;
+ this.responseCursor = responseCursor;
+ }
+
+
+ /**
+ * If the user requests our cursor, return the cursor
+ * @return
+ */
+ public Optional<String> getCursor(){
+ return this.responseCursor.encodeAsString();
+ }
+
+ public R getResult(){
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/CursorSerializerUtil.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/CursorSerializerUtil.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/CursorSerializerUtil.java
index 05c1018..fea0364 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/CursorSerializerUtil.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/CursorSerializerUtil.java
@@ -35,15 +35,9 @@ public class CursorSerializerUtil {
private static final ObjectMapper MAPPER = new ObjectMapper( SMILE_FACTORY );
- private static final Base64Variant VARIANT = Base64Variants.MODIFIED_FOR_URL;
-
public static ObjectMapper getMapper() {
return MAPPER;
}
-
- public static Base64Variant getBase64() {
- return VARIANT;
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/NoCursorSerializer.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/NoCursorSerializer.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/NoCursorSerializer.java
deleted file mode 100644
index 1d42df4..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/NoCursorSerializer.java
+++ /dev/null
@@ -1,55 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.pipeline.cursor;
-
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-
-/**
- * Interface for cursor serialization
- *
- * TODO, the need for this seems to indicate an issue with our object composition. Refactor this away
- */
-public class NoCursorSerializer<T> implements CursorSerializer<T> {
-
- private static final NoCursorSerializer<Object> INSTANCE = new NoCursorSerializer<>();
-
-
- @Override
- public T fromJsonNode( final JsonNode node, final ObjectMapper objectMapper ) {
- return null;
- }
-
-
- @Override
- public JsonNode toNode( final ObjectMapper objectMapper, final T value ) {
- return objectMapper.createObjectNode();
- }
-
-
- /**
- * convenience for type casting
- */
- public static <T> NoCursorSerializer<T> create() {
- return ( NoCursorSerializer<T> ) INSTANCE;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/RequestCursor.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/RequestCursor.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/RequestCursor.java
index b117c21..9da0a8a 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/RequestCursor.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/RequestCursor.java
@@ -45,7 +45,6 @@ public class RequestCursor {
private static final int MAX_CURSOR_COUNT = 100;
private static final ObjectMapper MAPPER = CursorSerializerUtil.getMapper();
- private static final Base64Variant VARIANT = CursorSerializerUtil.getBase64();
private final Map<Integer, JsonNode> parsedCursor;
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/ResponseCursor.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/ResponseCursor.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/ResponseCursor.java
index e379a34..f1c8c24 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/ResponseCursor.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/cursor/ResponseCursor.java
@@ -30,6 +30,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Optional;
/**
@@ -39,7 +40,6 @@ public class ResponseCursor {
private static final ObjectMapper MAPPER = CursorSerializerUtil.getMapper();
- private static final Base64Variant VARIANT = CursorSerializerUtil.getBase64();
/**
* We use a map b/c some indexes might be skipped
@@ -61,8 +61,14 @@ public class ResponseCursor {
/**
* now we're done, encode as a string
*/
- public String encodeAsString() {
+ public Optional<String> encodeAsString() {
try {
+
+ if(cursors.isEmpty()){
+ return Optional.absent();
+ }
+
+
final ObjectNode map = MAPPER.createObjectNode();
for ( Map.Entry<Integer, CursorEntry<?>> entry : cursors.entrySet() ) {
@@ -78,8 +84,9 @@ public class ResponseCursor {
final byte[] output = MAPPER.writeValueAsBytes(map);
//generate a base64 url save string
- return Base64.getUrlEncoder().encodeToString( output );
-// return MAPPER.writer( VARIANT ).writeValueAsString( map );
+ final String value = Base64.getUrlEncoder().encodeToString( output );
+
+ return Optional.of( value );
}
catch ( JsonProcessingException e ) {
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractFilter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractFilter.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractFilter.java
deleted file mode 100644
index 13ab185..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractFilter.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.pipeline.read;
-
-
-import java.io.Serializable;
-
-import org.apache.usergrid.corepersistence.pipeline.cursor.CursorSerializer;
-import org.apache.usergrid.corepersistence.pipeline.cursor.RequestCursor;
-import org.apache.usergrid.corepersistence.pipeline.cursor.ResponseCursor;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import com.google.common.base.Optional;
-
-
-/**
- * Basic functionality for our commands to handle cursor IO
- * @param <T> the input type
- * @param <R> The output Type
- * @param <C> The cursor Type
- */
-public abstract class AbstractFilter<T, R, C extends Serializable> implements Filter<T, R> {
-
- private int id;
- /**
- * The cache of the cursor that was set when the read was started
- */
- private RequestCursor readCache;
-
- /**
- * The current state of the write cache. Gets updated as we traverse the observables
- */
- private ResponseCursor writeCache;
-
-
- /**
- * The applicationScope
- */
- protected ApplicationScope applicationScope;
-
-
- @Override
- public void setId( final int id ) {
- this.id = id;
- }
-
-
- @Override
- public void setCursorCaches( final RequestCursor readCache, final ResponseCursor writeCache ) {
- this.readCache = readCache;
- this.writeCache = writeCache;
- }
-
-
- @Override
- public void setApplicationScope( final ApplicationScope applicationScope ) {
- this.applicationScope = applicationScope;
- }
-
-
- /**
- * Return the parsed value of the cursor from the last request, if it exists
- */
- protected Optional<C> getCursor() {
- final C cursor = readCache.getCursor( id, getCursorSerializer() );
-
- return Optional.fromNullable( cursor );
- }
-
-
-
-
-
- /**
- * Set the cursor value into the new cursor write cache
- * @param newValue
- */
- protected void setCursor(final C newValue){
- writeCache.setCursor( id, newValue, getCursorSerializer() );
- }
-
-
- /**
- * Generate our state as a cursor
- * @return
- */
- protected String generateCursor(){
- return writeCache.encodeAsString();
- }
-
- /**
- * Return the class to be used when parsing the cursor
- */
- protected abstract CursorSerializer<C> getCursorSerializer();
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractPipelineOperation.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractPipelineOperation.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractPipelineOperation.java
new file mode 100644
index 0000000..8d7f106
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractPipelineOperation.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.pipeline.read;
+
+
+import org.apache.usergrid.corepersistence.pipeline.PipelineContext;
+
+
+/**
+ * Basic functionality for our commands to handle cursor IO
+ * @param <T> the input type
+ * @param <R> The output Type
+ */
+public abstract class AbstractPipelineOperation<T, R> implements PipelineOperation<T, R> {
+
+
+ protected PipelineContext pipelineContext;
+
+
+ @Override
+ public void setContext( final PipelineContext pipelineContext ) {
+ this.pipelineContext = pipelineContext;
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractResumableFilter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractResumableFilter.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractResumableFilter.java
new file mode 100644
index 0000000..fbb6367
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractResumableFilter.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.pipeline.read;
+
+
+import java.io.Serializable;
+
+import org.apache.usergrid.corepersistence.pipeline.cursor.CursorSerializer;
+
+import com.google.common.base.Optional;
+
+
+/**
+ * Abstract class for filters to extend that require a cursor
+ * @param <T> The input type
+ * @param <R> The response type
+ * @param <C> The cursor type
+ */
+public abstract class AbstractResumableFilter<T, R, C extends Serializable> extends AbstractPipelineOperation<T, R> implements Filter<T, R> {
+
+
+
+
+ /**
+ * Return the parsed value of the cursor from the last request, if it exists
+ */
+ protected Optional<C> getCursor() {
+ return pipelineContext.getCursor( getCursorSerializer() );
+ }
+
+
+ /**
+ * Sets the cursor into our pipeline context
+ * @param newValue
+ */
+ protected void setCursor(final C newValue){
+ pipelineContext.setCursorValue( newValue, getCursorSerializer() );
+ }
+
+
+ /**
+ * Return the class to be used when parsing the cursor
+ */
+ protected abstract CursorSerializer<C> getCursorSerializer();
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CandidateResultsFilter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CandidateResultsFilter.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CandidateResultsFilter.java
index a5563b7..4e6d06e 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CandidateResultsFilter.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CandidateResultsFilter.java
@@ -28,4 +28,4 @@ import org.apache.usergrid.persistence.model.entity.Id;
* Traverses edges in the graph. Either by query or graph traversal. Take an observable of ids, and emits
* an observable of ids
*/
-public interface CandidateResultsFilter extends Filter<Id, CandidateResults> {}
+public interface CandidateResultsFilter extends PipelineOperation<Id, CandidateResults> {}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Collector.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Collector.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Collector.java
index cafd52a..69d929c 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Collector.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Collector.java
@@ -24,13 +24,8 @@ package org.apache.usergrid.corepersistence.pipeline.read;
* A command that is used to reduce our stream of results into a final output
* @param <T>
*/
-public interface Collector<T, R> extends Filter<T, R> {
+public interface Collector<T, R> extends PipelineOperation<T, R> {
- /**
- * Set the prefered result size for the command
- * @param limit
- */
- void setLimit( final int limit );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CollectorFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CollectorFactory.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CollectorFactory.java
new file mode 100644
index 0000000..6893b34
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CollectorFactory.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.pipeline.read;
+
+
+import org.apache.usergrid.corepersistence.pipeline.read.elasticsearch.CandidateResultsEntityResultsCollector;
+import org.apache.usergrid.corepersistence.pipeline.read.entity.EntityLoadCollector;
+
+
+/**
+ * A factory for generating collectors
+ */
+public interface CollectorFactory {
+
+ /**
+ * Generate a new instance of the command with the specified parameters
+ */
+ EntityLoadCollector entityLoadCollector();
+
+ /**
+ * Get the collector for collection candidate results to entities
+ * @return
+ */
+ CandidateResultsEntityResultsCollector candidateResultsEntityResultsCollector();
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Filter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Filter.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Filter.java
index 09b2e2b..ace62db 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Filter.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/Filter.java
@@ -20,40 +20,11 @@
package org.apache.usergrid.corepersistence.pipeline.read;
-import org.apache.usergrid.corepersistence.pipeline.cursor.RequestCursor;
-import org.apache.usergrid.corepersistence.pipeline.cursor.ResponseCursor;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Id;
-import rx.Observable;
-
/**
- * Interface for filtering commands. All filters must take an observable of Id's as an input. Output is then determined by subclasses.
- * This takes an input of Id, performs some operation, and emits values for further processing in the Observable
- * pipeline
- * @param <T> The input type
- * @param <R>
+ * Traverses edges in the graph. Either by query or graph traversal. Take an observable of ids, and emits
+ * an observable of ids
*/
-public interface Filter< T, R> extends Observable.Transformer<T, R> {
-
-
- /**
- * Set the id of this filter in it's execution environment
- */
- void setId( final int id );
-
- /**
- * Set the cursor cache into the filter
- *
- * @param readCache Set the cache that was used in the request
- * @param writeCache Set the cache to be used when writing the results
- */
- void setCursorCaches( final RequestCursor readCache, final ResponseCursor writeCache );
-
- /**
- * Set the application scope of the filter
- * @param applicationScope
- */
- void setApplicationScope(final ApplicationScope applicationScope);
-}
+public interface Filter<T, R> extends PipelineOperation<T, R> {}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/FilterFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/FilterFactory.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/FilterFactory.java
new file mode 100644
index 0000000..26a3dfa
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/FilterFactory.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.pipeline.read;
+
+
+import org.apache.usergrid.corepersistence.pipeline.read.elasticsearch.CandidateResultsIdVerifyFilter;
+import org.apache.usergrid.corepersistence.pipeline.read.elasticsearch.CollectionElasticSearchFilter;
+import org.apache.usergrid.corepersistence.pipeline.read.elasticsearch.ConnectionElasticSearchFilter;
+import org.apache.usergrid.corepersistence.pipeline.read.entity.EntityIdFilter;
+import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphCollectionByIdFilter;
+import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphCollectionFilter;
+import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphConnectionByIdFilter;
+import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphConnectionByTypeFilter;
+import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphConnectionFilter;
+import org.apache.usergrid.persistence.model.entity.Id;
+
+import com.google.common.base.Optional;
+
+
+/**
+ * A factory for generating read commands
+ */
+public interface FilterFactory {
+
+
+ /**
+ * Generate a new instance of the command with the specified parameters
+ */
+ ReadGraphCollectionFilter readGraphCollectionFilter( final String collectionName );
+
+ /**
+ * Read a connection between two entities, the incoming and the target entity
+ * @param collectionName
+ * @param targetId
+ * @return
+ */
+ ReadGraphCollectionByIdFilter readGraphCollectionByIdFilter(final String collectionName, final Id targetId);
+
+ /**
+ * Generate a new instance of the command with the specified parameters
+ */
+ ReadGraphConnectionFilter readGraphConnectionFilter( final String connectionName );
+
+ /**
+ * Generate a new instance of the command with the specified parameters
+ */
+ ReadGraphConnectionByTypeFilter readGraphConnectionByTypeFilter( final String connectionName, final String entityType );
+
+
+ /**
+ * Read a connection directly between two identifiers
+ * @param connectionName
+ * @param targetId
+ * @return
+ */
+ ReadGraphConnectionByIdFilter readGraphConnectionByIdFilter(final String connectionName, final Id targetId);
+
+ /**
+ * Generate a new instance of the command with the specified parameters
+ */
+ CollectionElasticSearchFilter collectionElasticSearchFilter(final String query, final String collectionName );
+
+
+ /**
+ * Generate a new instance of the command with the specified parameters
+ */
+ ConnectionElasticSearchFilter connectionElasticSearchFilter( final String query, final String connectionName, final Optional<String> connectionType);
+
+
+ /**
+ * Get a candidate ids verifier for collection results
+ * @return
+ */
+ CandidateResultsIdVerifyFilter candidateResultsIdVerifyFilter();
+
+ /**
+ * Get an entity id filter. Used as a 1.0->2.0 bridge since we're not doing full traversals
+ */
+ EntityIdFilter getEntityIdFilter( final Id entityId );
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/IdFilter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/IdFilter.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/IdFilter.java
deleted file mode 100644
index c059ed4..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/IdFilter.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.pipeline.read;
-
-
-import org.apache.usergrid.persistence.model.entity.Id;
-
-
-/**
- * Traverses edges in the graph. Either by query or graph traversal. Take an observable of ids, and emits
- * an observable of ids
- */
-public interface IdFilter extends Filter<Id, Id> {}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/PipelineOperation.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/PipelineOperation.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/PipelineOperation.java
new file mode 100644
index 0000000..28bba36
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/PipelineOperation.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.pipeline.read;
+
+
+import org.apache.usergrid.corepersistence.pipeline.PipelineContext;
+
+import rx.Observable;
+
+
+/**
+ * Interface for filtering commands. All filters must take an observable of Id's as an input. Output is then determined by subclasses.
+ * This takes an input of Id, performs some operation, and emits values for further processing in the Observable
+ * pipeline
+ * @param <T> The input type
+ * @param <R>
+ */
+public interface PipelineOperation< T, R> extends Observable.Transformer<T, R> {
+
+ void setContext(final PipelineContext pipelineContext);
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadFilterFactory.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadFilterFactory.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadFilterFactory.java
deleted file mode 100644
index d4754ed..0000000
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadFilterFactory.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.usergrid.corepersistence.pipeline.read;
-
-
-import org.apache.usergrid.corepersistence.pipeline.read.elasticsearch.QueryCollectionElasticSearchCollectorFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.elasticsearch.QueryConnectionElasticSearchCollectorFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.entity.EntityIdFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.entity.EntityLoadCollector;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphCollectionByIdFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphCollectionFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphConnectionByIdFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphConnectionByTypeFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphConnectionFilter;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-
-/**
- * A factory for generating read commands
- */
-public interface ReadFilterFactory {
-
-
- /**
- * Generate a new instance of the command with the specified parameters
- */
- ReadGraphCollectionFilter readGraphCollectionCommand( final String collectionName );
-
- /**
- * Read a connection between two entities, the incoming and the target entity
- * @param collectionName
- * @param targetId
- * @return
- */
- ReadGraphCollectionByIdFilter readGraphCollectionByIdFilter(final String collectionName, final Id targetId);
-
- /**
- * Generate a new instance of the command with the specified parameters
- */
- ReadGraphConnectionFilter readGraphConnectionCommand( final String connectionName );
-
- /**
- * Generate a new instance of the command with the specified parameters
- */
- ReadGraphConnectionByTypeFilter readGraphConnectionCommand( final String connectionName, final String entityType );
-
-
- /**
- * Read a connection directly between two identifiers
- * @param connectionName
- * @param targetId
- * @return
- */
- ReadGraphConnectionByIdFilter readGraphConnectionByIdFilter(final String connectionName, final Id targetId);
-
- /**
- * Generate a new instance of the command with the specified parameters
- */
- EntityLoadCollector entityLoadCollector();
-
- /**
- * Generate a new instance of the command with the specified parameters
- */
- QueryCollectionElasticSearchCollectorFilter queryCollectionElasticSearchCollector( final String collectionName, final String query);
-
-
- /**
- * Generate a new instance of the command with the specified parameters
- */
- QueryConnectionElasticSearchCollectorFilter queryConnectionElasticSearchCollector( final String connectionName,final String query);
-
-
- /**
- * Generate a new instance of the command with the specified parameters
- */
- QueryConnectionElasticSearchCollectorFilter queryConnectionElasticSearchCollector( final String connectionName, final String connectionEntityType, final String query );
-
-
- /**
- * Get an entity id filter. Used as a 1.0->2.0 bridge since we're not doing full traversals
- */
- EntityIdFilter getEntityIdFilter( final Id entityId );
-}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadFilterFactoryImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadFilterFactoryImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadFilterFactoryImpl.java
index e3d15e6..0f73fb9 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadFilterFactoryImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadFilterFactoryImpl.java
@@ -20,133 +20,117 @@
package org.apache.usergrid.corepersistence.pipeline.read;
-import org.apache.usergrid.corepersistence.pipeline.read.elasticsearch.QueryCollectionElasticSearchCollectorFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.elasticsearch.QueryConnectionElasticSearchCollectorFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.entity.EntityIdFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.entity.EntityLoadCollector;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphCollectionByIdFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphCollectionFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphConnectionByIdFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphConnectionByTypeFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.graph.ReadGraphConnectionFilter;
-import org.apache.usergrid.persistence.Query;
-import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory;
-import org.apache.usergrid.persistence.graph.GraphManagerFactory;
-import org.apache.usergrid.persistence.index.EntityIndexFactory;
-import org.apache.usergrid.persistence.model.entity.Id;
-
-import com.google.inject.Inject;
import com.google.inject.Singleton;
@Singleton
-public class ReadFilterFactoryImpl implements ReadFilterFactory {
-
-
- private final GraphManagerFactory graphManagerFactory;
- private final EntityIndexFactory entityIndexFactory;
- private final EntityCollectionManagerFactory entityCollectionManagerFactory;
-
-
- @Inject
- public ReadFilterFactoryImpl( final GraphManagerFactory graphManagerFactory,
- final EntityIndexFactory entityIndexFactory,
- final EntityCollectionManagerFactory entityCollectionManagerFactory ) {
-
-
- this.graphManagerFactory = graphManagerFactory;
- this.entityIndexFactory = entityIndexFactory;
- this.entityCollectionManagerFactory = entityCollectionManagerFactory;
- }
-
-
- @Override
- public ReadGraphCollectionFilter readGraphCollectionCommand( final String collectionName ) {
- return new ReadGraphCollectionFilter( graphManagerFactory, collectionName );
- }
-
-
- @Override
- public ReadGraphCollectionByIdFilter readGraphCollectionByIdFilter( final String collectionName,
- final Id targetId ) {
- return new ReadGraphCollectionByIdFilter( graphManagerFactory, collectionName, targetId );
- }
-
-
- @Override
- public ReadGraphConnectionFilter readGraphConnectionCommand( final String connectionName ) {
- return new ReadGraphConnectionFilter( graphManagerFactory, connectionName );
- }
-
-
- @Override
- public ReadGraphConnectionByTypeFilter readGraphConnectionCommand( final String connectionName,
- final String entityType ) {
- return new ReadGraphConnectionByTypeFilter( graphManagerFactory, connectionName, entityType );
- }
-
-
- @Override
- public ReadGraphConnectionByIdFilter readGraphConnectionByIdFilter( final String connectionName,
- final Id targetId ) {
- return new ReadGraphConnectionByIdFilter( graphManagerFactory, connectionName, targetId );
- }
-
-
- @Override
- public EntityLoadCollector entityLoadCollector() {
- return new EntityLoadCollector( entityCollectionManagerFactory );
- }
-
-
- /**
- * TODO refactor these impls to use RX internally, as well as remove the query object
- */
- @Override
- public QueryCollectionElasticSearchCollectorFilter queryCollectionElasticSearchCollector(
- final String collectionName, final String query ) {
-
- final Query queryObject = Query.fromQL( query );
-
- final QueryCollectionElasticSearchCollectorFilter filter =
- new QueryCollectionElasticSearchCollectorFilter( entityCollectionManagerFactory, entityIndexFactory,
- collectionName, queryObject );
-
- return filter;
- }
-
-
- @Override
- public QueryConnectionElasticSearchCollectorFilter queryConnectionElasticSearchCollector(
- final String connectionName, final String query ) {
-
- final Query queryObject = Query.fromQL( query );
-
- final QueryConnectionElasticSearchCollectorFilter filter =
- new QueryConnectionElasticSearchCollectorFilter( entityCollectionManagerFactory, entityIndexFactory,
- connectionName, queryObject );
-
- return filter;
- }
-
-
- @Override
- public QueryConnectionElasticSearchCollectorFilter queryConnectionElasticSearchCollector(
- final String connectionName, final String connectionEntityType, final String query ) {
-
- final Query queryObject = Query.fromQL( query );
- queryObject.setConnectionType( connectionEntityType );
-
- final QueryConnectionElasticSearchCollectorFilter filter =
- new QueryConnectionElasticSearchCollectorFilter( entityCollectionManagerFactory, entityIndexFactory,
- connectionName, queryObject );
-
- return filter;
- }
-
-
- @Override
- public EntityIdFilter getEntityIdFilter( final Id entityId ) {
- return new EntityIdFilter( entityId );
- }
+public class ReadFilterFactoryImpl { //implements ReadFilterFactory {
+
+//
+// private final GraphManagerFactory graphManagerFactory;
+// private final EntityIndexFactory entityIndexFactory;
+// private final EntityCollectionManagerFactory entityCollectionManagerFactory;
+//
+//
+// @Inject
+// public ReadFilterFactoryImpl( final GraphManagerFactory graphManagerFactory,
+// final EntityIndexFactory entityIndexFactory,
+// final EntityCollectionManagerFactory entityCollectionManagerFactory ) {
+//
+//
+// this.graphManagerFactory = graphManagerFactory;
+// this.entityIndexFactory = entityIndexFactory;
+// this.entityCollectionManagerFactory = entityCollectionManagerFactory;
+// }
+//
+//
+// @Override
+// public ReadGraphCollectionFilter readGraphCollectionCommand( final String collectionName ) {
+// return new ReadGraphCollectionFilter( graphManagerFactory, collectionName );
+// }
+//
+//
+// @Override
+// public ReadGraphCollectionByIdFilter readGraphCollectionByIdFilter( final String collectionName,
+// final Id targetId ) {
+// return new ReadGraphCollectionByIdFilter( graphManagerFactory, collectionName, targetId );
+// }
+//
+//
+// @Override
+// public ReadGraphConnectionFilter readGraphConnectionCommand( final String connectionName ) {
+// return new ReadGraphConnectionFilter( graphManagerFactory, connectionName );
+// }
+//
+//
+// @Override
+// public ReadGraphConnectionByTypeFilter readGraphConnectionCommand( final String connectionName,
+// final String entityType ) {
+// return new ReadGraphConnectionByTypeFilter( graphManagerFactory, connectionName, entityType );
+// }
+//
+//
+// @Override
+// public ReadGraphConnectionByIdFilter readGraphConnectionByIdFilter( final String connectionName,
+// final Id targetId ) {
+// return new ReadGraphConnectionByIdFilter( graphManagerFactory, connectionName, targetId );
+// }
+//
+//
+// @Override
+// public EntityLoadCollector entityLoadCollector() {
+// return new EntityLoadCollector( entityCollectionManagerFactory );
+// }
+//
+//
+// /**
+// * TODO refactor these impls to use RX internally, as well as remove the query object
+// */
+// @Override
+// public QueryCollectionElasticSearchCollectorFilter queryCollectionElasticSearchCollector(
+// final String collectionName, final String query ) {
+//
+// final Query queryObject = Query.fromQL( query );
+//
+// final QueryCollectionElasticSearchCollectorFilter filter =
+// new QueryCollectionElasticSearchCollectorFilter( entityCollectionManagerFactory, entityIndexFactory,
+// collectionName, queryObject );
+//
+// return filter;
+// }
+//
+//
+// @Override
+// public QueryConnectionElasticSearchCollectorFilter queryConnectionElasticSearchCollector(
+// final String connectionName, final String query ) {
+//
+// final Query queryObject = Query.fromQL( query );
+//
+// final QueryConnectionElasticSearchCollectorFilter filter =
+// new QueryConnectionElasticSearchCollectorFilter( entityCollectionManagerFactory, entityIndexFactory,
+// connectionName, queryObject );
+//
+// return filter;
+// }
+//
+//
+// @Override
+// public QueryConnectionElasticSearchCollectorFilter queryConnectionElasticSearchCollector(
+// final String connectionName, final String connectionEntityType, final String query ) {
+//
+// final Query queryObject = Query.fromQL( query );
+// queryObject.setConnectionType( connectionEntityType );
+//
+// final QueryConnectionElasticSearchCollectorFilter filter =
+// new QueryConnectionElasticSearchCollectorFilter( entityCollectionManagerFactory, entityIndexFactory,
+// connectionName, queryObject );
+//
+// return filter;
+// }
+//
+//
+// @Override
+// public EntityIdFilter getEntityIdFilter( final Id entityId ) {
+// return new EntityIdFilter( entityId );
+// }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilder.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilder.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilder.java
index 5035773..9da2b03 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilder.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilder.java
@@ -20,6 +20,8 @@
package org.apache.usergrid.corepersistence.pipeline.read;
+import org.apache.usergrid.corepersistence.pipeline.PipelineResult;
+import org.apache.usergrid.persistence.Entity;
import org.apache.usergrid.persistence.Results;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -29,8 +31,10 @@ import rx.Observable;
/**
- * An instance of a pipeline builder for building commands.
- * Each invocation of the method will assemple the underlying pipe and updating it's state
+ * An instance of a pipeline builder for building commands on our read pipline
+ *
+ * Each invocation of the method will assemble the underlying pipe and updating it's state
+ *
* Results are added by invoking execute.
*/
public interface ReadPipelineBuilder {
@@ -89,24 +93,14 @@ public interface ReadPipelineBuilder {
ReadPipelineBuilder getConnection( final String connectionName, final String entityType );
/**
- * Get all entities in a connection with a query
- */
- ReadPipelineBuilder connectionWithQuery( final String connectionName, final String query );
-
-
- /**
* Get all entities in a connection with a query and a target entity type
*/
- ReadPipelineBuilder connectionWithQuery( final String connectionName, final String entityType, final String query);
-
-
-
-
+ ReadPipelineBuilder connectionWithQuery( final String connectionName, final Optional<String> entityType, final String query);
/**
- * Execute final construction of the pipeline and return the results
+ * Load our entity results when our previous filter calls graph
* @return
*/
- Observable<Results> build();
+ Observable<PipelineResult<ResultsPage>> execute();
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilderImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilderImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilderImpl.java
index f5780a0..8ec4f42 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilderImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ReadPipelineBuilderImpl.java
@@ -20,8 +20,12 @@
package org.apache.usergrid.corepersistence.pipeline.read;
-import org.apache.usergrid.corepersistence.pipeline.DataPipeline;
-import org.apache.usergrid.persistence.Results;
+import java.util.List;
+
+import org.apache.usergrid.corepersistence.pipeline.Pipeline;
+import org.apache.usergrid.corepersistence.pipeline.PipelineResult;
+import org.apache.usergrid.corepersistence.pipeline.read.elasticsearch.CandidateResultsEntityResultsCollector;
+import org.apache.usergrid.corepersistence.pipeline.read.entity.EntityLoadCollector;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.model.entity.Id;
@@ -40,59 +44,62 @@ public class ReadPipelineBuilderImpl implements ReadPipelineBuilder {
private static final int DEFAULT_LIMIT = 10;
- private final ReadFilterFactory readFilterFactory;
+ private final FilterFactory filterFactory;
+
+ private final CollectorState collectorState;
+
+ private final ApplicationScope applicationScope;
- private final DataPipeline pipeline;
/**
* Our pointer to our collect filter. Set or cleared with each operation that's performed so the correct results are
* rendered
*/
- private Collector<Results> collector;
+ private List<Filter> filters;
+
private Optional<String> cursor;
private int limit;
@Inject
- public ReadPipelineBuilderImpl( final ReadFilterFactory readFilterFactory,
+ public ReadPipelineBuilderImpl( final FilterFactory filterFactory, final CollectorFactory collectorFactory,
@Assisted final ApplicationScope applicationScope ) {
- this.readFilterFactory = readFilterFactory;
+ this.filterFactory = filterFactory;
- //set up our pipeline with our application scope
- this.pipeline = new DataPipeline( applicationScope );
//init our cursor to empty
this.cursor = Optional.absent();
//set the default limit
this.limit = DEFAULT_LIMIT;
+
+ collectorState = new CollectorState( collectorFactory );
+
+ this.applicationScope = applicationScope;
}
@Override
public ReadPipelineBuilder withCursor( final Optional<String> cursor ) {
- this.cursor = cursor ;
- pipeline.setCursor( this.cursor );
+ this.cursor = cursor;
return this;
}
@Override
public ReadPipelineBuilder withLimit( final Optional<Integer> limit ) {
-
- this. limit = limit.or( DEFAULT_LIMIT );
- //set the default value
- pipeline.setLimit( this.limit);
+ this.limit = limit.or( DEFAULT_LIMIT );
return this;
}
@Override
public ReadPipelineBuilder setStartId( final Id id ) {
- pipeline.withTraverseCommand( readFilterFactory.getEntityIdFilter( id ) );
+ filters.add( filterFactory.getEntityIdFilter( id ) );
- this.collector = null;
+ this.collectorState.clear();
+ ;
return this;
@@ -102,9 +109,9 @@ public class ReadPipelineBuilderImpl implements ReadPipelineBuilder {
@Override
public ReadPipelineBuilder getEntityViaCollection( final String collectionName, final Id entityId ) {
- pipeline.withTraverseCommand( readFilterFactory.readGraphCollectionByIdFilter( collectionName, entityId ) );
+ filters.add( filterFactory.readGraphCollectionByIdFilter( collectionName, entityId ) );
- setEntityLoaderFilter();
+ this.collectorState.setEntityLoaderCollector();
return this;
}
@@ -114,9 +121,9 @@ public class ReadPipelineBuilderImpl implements ReadPipelineBuilder {
public ReadPipelineBuilder getCollection( final String collectionName ) {
- pipeline.withTraverseCommand( readFilterFactory.readGraphCollectionCommand( collectionName ) );
+ filters.add( filterFactory.readGraphCollectionFilter( collectionName ) );
- setEntityLoaderFilter();
+ this.collectorState.setEntityLoaderCollector();
return this;
}
@@ -126,7 +133,11 @@ public class ReadPipelineBuilderImpl implements ReadPipelineBuilder {
public ReadPipelineBuilder getCollectionWithQuery( final String collectionName, final String query ) {
//TODO, this should really be 2 a TraverseFilter with an entityLoad collector
- collector = readFilterFactory.queryCollectionElasticSearchCollector( collectionName, query );
+
+ filters.add( filterFactory.collectionElasticSearchFilter( query, collectionName ) );
+
+ this.collectorState.setCandidateResultsEntityResultsCollector();
+
return this;
}
@@ -134,8 +145,8 @@ public class ReadPipelineBuilderImpl implements ReadPipelineBuilder {
@Override
public ReadPipelineBuilder getEntityViaConnection( final String connectionName, final Id entityId ) {
- pipeline.withTraverseCommand( readFilterFactory.readGraphConnectionByIdFilter( connectionName, entityId ) );
- setEntityLoaderFilter();
+ filters.add( filterFactory.readGraphConnectionByIdFilter( connectionName, entityId ) );
+ collectorState.setEntityLoaderCollector();
return this;
}
@@ -144,8 +155,8 @@ public class ReadPipelineBuilderImpl implements ReadPipelineBuilder {
@Override
public ReadPipelineBuilder getConnection( final String connectionName ) {
- pipeline.withTraverseCommand( readFilterFactory.readGraphConnectionCommand( connectionName ) );
- setEntityLoaderFilter();
+ filters.add( filterFactory.readGraphConnectionFilter( connectionName ) );
+ collectorState.setEntityLoaderCollector();
return this;
}
@@ -153,51 +164,82 @@ public class ReadPipelineBuilderImpl implements ReadPipelineBuilder {
@Override
public ReadPipelineBuilder getConnection( final String connectionName, final String entityType ) {
- pipeline.withTraverseCommand( readFilterFactory.readGraphConnectionCommand( connectionName, entityType ) );
- setEntityLoaderFilter();
+ filters.add( filterFactory.readGraphConnectionByTypeFilter( connectionName, entityType ) );
+ collectorState.setEntityLoaderCollector();
return this;
}
- /**
- *
- * @param connectionName
- * @param query
- * @return
- */
@Override
- public ReadPipelineBuilder connectionWithQuery( final String connectionName, final String query ) {
-
- //TODO, this should really be 2 a TraverseFilter with an entityLoad collector
- collector = readFilterFactory.queryConnectionElasticSearchCollector( connectionName, query );
-
- return this;
- }
-
-
- @Override
- public ReadPipelineBuilder connectionWithQuery( final String connectionName, final String entityType,
+ public ReadPipelineBuilder connectionWithQuery( final String connectionName, final Optional<String> entityType,
final String query ) {
-
- //TODO, this should really be 2 a TraverseFilter with an entityLoad collector
- collector =
- readFilterFactory.queryConnectionElasticSearchCollector( connectionName, entityType, query);
+ filters.add( filterFactory.connectionElasticSearchFilter( query, connectionName, entityType ) );
+ collectorState.setCandidateResultsEntityResultsCollector();
return this;
}
@Override
- public Observable<Results> build() {
+ public Observable<PipelineResult<ResultsPage>> execute() {
+ final Collector<?, ResultsPage> collector = collectorState.getCollector();
+
Preconditions.checkNotNull( collector,
"You have not specified an operation that creates a collection filter. This is required for loading "
+ "results" );
- return pipeline.build( collector );
+
+ Pipeline pipeline = new Pipeline( applicationScope, filters, collector, cursor, limit );
+
+
+ return pipeline.execute();
}
- private void setEntityLoaderFilter() {
- collector = readFilterFactory.entityLoadCollector();
+ /**
+ * A mutable state for our collectors. Rather than create a new instance each time, we create a singleton
+ * collector
+ */
+ private static final class CollectorState {
+ private final CollectorFactory collectorFactory;
+
+ private EntityLoadCollector entityLoadCollector;
+
+ private CandidateResultsEntityResultsCollector candidateResultsEntityResultsCollector;
+
+
+ private Collector<?, ResultsPage> collector = null;
+
+
+ private CollectorState( final CollectorFactory collectorFactory ) {this.collectorFactory = collectorFactory;}
+
+
+ public void setEntityLoaderCollector() {
+ if ( entityLoadCollector == null ) {
+ entityLoadCollector = collectorFactory.entityLoadCollector();
+ }
+
+
+ collector = entityLoadCollector;
+ }
+
+
+ public void setCandidateResultsEntityResultsCollector() {
+ if ( candidateResultsEntityResultsCollector == null ) {
+ candidateResultsEntityResultsCollector = collectorFactory.candidateResultsEntityResultsCollector();
+ }
+
+ collector = candidateResultsEntityResultsCollector;
+ }
+
+
+ public void clear() {
+ collector = null;
+ }
+
+
+ public Collector<?, ResultsPage> getCollector() {
+ return collector;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ResultsPage.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ResultsPage.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ResultsPage.java
new file mode 100644
index 0000000..e428e7a
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/ResultsPage.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.usergrid.corepersistence.pipeline.read;
+
+
+import java.util.List;
+
+import org.apache.usergrid.persistence.model.entity.Entity;
+
+
+/**
+ * An encapsulation of entities as a group of responses. Ordered by the requesting filters. Each set should be considered a "page" of results.
+ */
+public class ResultsPage {
+
+ private final List<Entity> entityList;
+
+
+ public ResultsPage( final List<Entity> entityList ) {this.entityList = entityList;}
+
+
+ public List<Entity> getEntityList() {
+ return entityList;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/d8c0f811/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/elasticsearch/AbstractElasticSearchFilter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/elasticsearch/AbstractElasticSearchFilter.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/elasticsearch/AbstractElasticSearchFilter.java
index 2d4d991..83c394e 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/elasticsearch/AbstractElasticSearchFilter.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/elasticsearch/AbstractElasticSearchFilter.java
@@ -24,9 +24,9 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.usergrid.corepersistence.pipeline.cursor.CursorSerializer;
-import org.apache.usergrid.corepersistence.pipeline.read.AbstractFilter;
+import org.apache.usergrid.corepersistence.pipeline.read.AbstractPipelineOperation;
+import org.apache.usergrid.corepersistence.pipeline.read.AbstractResumableFilter;
import org.apache.usergrid.corepersistence.pipeline.read.CandidateResultsFilter;
-import org.apache.usergrid.corepersistence.pipeline.read.IdFilter;
import org.apache.usergrid.persistence.core.metrics.MetricsFactory;
import org.apache.usergrid.persistence.core.metrics.ObservableTimer;
import org.apache.usergrid.persistence.index.ApplicationEntityIndex;
@@ -45,25 +45,22 @@ import rx.Observable;
/**
* Command for reading graph edges
*/
-public abstract class AbstractElasticSearchFilter extends AbstractFilter<Id, CandidateResults, Integer>
+public abstract class AbstractElasticSearchFilter extends AbstractResumableFilter<Id, CandidateResults, Integer>
implements CandidateResultsFilter {
private static final Logger log = LoggerFactory.getLogger( AbstractElasticSearchFilter.class );
private final EntityIndexFactory entityIndexFactory;
private final String query;
- private final int limit;
private final Timer searchTimer;
/**
* Create a new instance of our command
*/
- public AbstractElasticSearchFilter( final EntityIndexFactory entityIndexFactory, final MetricsFactory metricsFactory, final String query,
- final int limit ) {
+ public AbstractElasticSearchFilter( final EntityIndexFactory entityIndexFactory, final MetricsFactory metricsFactory, final String query) {
this.entityIndexFactory = entityIndexFactory;
this.query = query;
- this.limit = limit;
this.searchTimer = metricsFactory.getTimer( AbstractElasticSearchFilter.class, "query" );
}
@@ -73,13 +70,15 @@ public abstract class AbstractElasticSearchFilter extends AbstractFilter<Id, Can
//get the graph manager
final ApplicationEntityIndex applicationEntityIndex =
- entityIndexFactory.createApplicationEntityIndex( applicationScope );
+ entityIndexFactory.createApplicationEntityIndex( pipelineContext.getApplicationScope() );
//set our our constant state
final Optional<Integer> startFromCursor = getCursor();
final int startOffset = startFromCursor.or( 0 );
+ final int limit = pipelineContext.getLimit();
+
final SearchTypes searchTypes = getSearchTypes();