You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by gr...@apache.org on 2015/05/26 20:13:08 UTC
[09/22] incubator-usergrid git commit: Finishes changes before tests
Finishes changes before tests
Project: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/commit/20c9b350
Tree: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/tree/20c9b350
Diff: http://git-wip-us.apache.org/repos/asf/incubator-usergrid/diff/20c9b350
Branch: refs/heads/USERGRID-608
Commit: 20c9b3509cf96a6ecab1a45a2c572fd6a041e00d
Parents: cb179d3
Author: Todd Nine <tn...@apigee.com>
Authored: Thu May 14 17:19:27 2015 -0600
Committer: Todd Nine <tn...@apigee.com>
Committed: Thu May 14 17:19:27 2015 -0600
----------------------------------------------------------------------
.../asyncevents/InMemoryAsyncEventService.java | 2 -
.../index/EdgeScopeSerializer.java | 41 ++++
.../index/IndexProcessorFig.java | 6 +-
.../corepersistence/index/ReIndexService.java | 75 +++---
.../index/ReIndexServiceImpl.java | 226 +++++++++++++++----
.../pipeline/cursor/CursorSerializerUtil.java | 54 ++++-
.../pipeline/cursor/RequestCursor.java | 9 +-
.../pipeline/cursor/ResponseCursor.java | 49 ++--
.../pipeline/read/AbstractPathFilter.java | 30 ---
.../pipeline/read/CursorSeek.java | 53 +++++
.../rx/impl/AllEntityIdsObservable.java | 4 +-
.../rx/impl/AllEntityIdsObservableImpl.java | 5 +-
.../PerformanceEntityRebuildIndexTest.java | 5 +-
.../graph/serialization/EdgesObservable.java | 21 +-
.../serialization/impl/EdgesObservableImpl.java | 4 +-
15 files changed, 422 insertions(+), 162 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
index 96966bf..ddcf826 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/asyncevents/InMemoryAsyncEventService.java
@@ -99,8 +99,6 @@ public class InMemoryAsyncEventService implements AsyncEventService {
@Override
public void index( final EntityIndexOperation entityIndexOperation ) {
-
-
run(eventBuilder.index( entityIndexOperation ));
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/EdgeScopeSerializer.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/EdgeScopeSerializer.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/EdgeScopeSerializer.java
new file mode 100644
index 0000000..2a6a5ac
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/EdgeScopeSerializer.java
@@ -0,0 +1,41 @@
+/*
+ * 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.index;
+
+
+import org.apache.usergrid.corepersistence.pipeline.cursor.AbstractCursorSerializer;
+import org.apache.usergrid.corepersistence.rx.impl.EdgeScope;
+
+
+/**
+ * Serialize our edge scope for cursors
+ */
+public class EdgeScopeSerializer extends AbstractCursorSerializer<EdgeScope> {
+
+
+ public static final EdgeScopeSerializer INSTANCE = new EdgeScopeSerializer();
+
+ @Override
+ protected Class<EdgeScope> getType() {
+ return EdgeScope.class;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
index fe9d3fd..8e835e2 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/IndexProcessorFig.java
@@ -78,9 +78,9 @@ public interface IndexProcessorFig extends GuicyFig {
String getQueueImplementation();
- @Default("30000")
- @Key("elasticsearch.reindex.sample.interval")
- long getReIndexSampleInterval();
+ @Default("10000")
+ @Key("elasticsearch.reindex.flush.interval")
+ int getUpdateInterval();
@Default("false")
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexService.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexService.java
index b25eca5..f8955dd 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexService.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexService.java
@@ -20,22 +20,6 @@
package org.apache.usergrid.corepersistence.index;
-import java.util.UUID;
-import java.util.concurrent.Callable;
-import java.util.concurrent.FutureTask;
-
-import org.apache.usergrid.corepersistence.rx.impl.AllEntityIdsObservable;
-import org.apache.usergrid.corepersistence.rx.impl.EdgeScope;
-import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
-import org.apache.usergrid.persistence.core.scope.ApplicationScope;
-
-import com.google.common.base.Optional;
-
-import rx.Observable;
-import rx.Observer;
-import rx.observables.ConnectableObservable;
-
-
/**
* An interface for re-indexing all entities in an application
*/
@@ -46,48 +30,75 @@ public interface ReIndexService {
* Perform an index rebuild
*
* @param indexServiceRequestBuilder The builder to build the request
- * @return
*/
- IndexResponse rebuildIndex(final IndexServiceRequestBuilder indexServiceRequestBuilder);
+ IndexResponse rebuildIndex( final IndexServiceRequestBuilder indexServiceRequestBuilder );
/**
* Generate a build for the index
- * @return
*/
IndexServiceRequestBuilder getBuilder();
+
+ /**
+ * Get the status of a job
+ * @param jobId The jobId returned during the rebuild index
+ * @return
+ */
+ IndexResponse getStatus( final String jobId );
+
+
/**
* The response when requesting a re-index operation
*/
class IndexResponse {
- final String cursor;
- final ConnectableObservable<EdgeScope> indexedEdgecount;
+ final String jobId;
+ final String status;
+ final long numberProcessed;
+ final long lastUpdated;
+
+
+ public IndexResponse( final String jobId, final String status, final long numberProcessed,
+ final long lastUpdated ) {
+ this.jobId = jobId;
+ this.status = status;
+ this.numberProcessed = numberProcessed;
+ this.lastUpdated = lastUpdated;
+ }
- public IndexResponse( final String cursor, final ConnectableObservable<EdgeScope> indexedEdgecount ) {
- this.cursor = cursor;
- this.indexedEdgecount = indexedEdgecount;
+ /**
+ * Get the jobId used to resume this operation
+ */
+ public String getJobId() {
+ return jobId;
+ }
+
+
+ /**
+ * Get the last updated time, as a long
+ * @return
+ */
+ public long getLastUpdated() {
+ return lastUpdated;
}
/**
- * Get the cursor used to resume this operation
+ * Get the number of records processed
* @return
*/
- public String getCursor() {
- return cursor;
+ public long getNumberProcessed() {
+ return numberProcessed;
}
/**
- * Return the observable of all edges to be indexed.
- *
- * Note that after subscribing "connect" will need to be called to ensure that processing begins
+ * Get the status
* @return
*/
- public ConnectableObservable<EdgeScope> getCount() {
- return indexedEdgecount;
+ public String getStatus() {
+ return status;
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
index a2fa09a..d828fc2 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/index/ReIndexServiceImpl.java
@@ -20,29 +20,32 @@
package org.apache.usergrid.corepersistence.index;
-import java.util.concurrent.TimeUnit;
+import java.util.List;
import org.apache.usergrid.corepersistence.asyncevents.AsyncEventService;
+import org.apache.usergrid.corepersistence.pipeline.cursor.CursorSerializerUtil;
+import org.apache.usergrid.corepersistence.pipeline.read.CursorSeek;
import org.apache.usergrid.corepersistence.rx.impl.AllApplicationsObservable;
import org.apache.usergrid.corepersistence.rx.impl.AllEntityIdsObservable;
import org.apache.usergrid.corepersistence.rx.impl.EdgeScope;
import org.apache.usergrid.corepersistence.util.CpNamingUtils;
-import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
-import org.apache.usergrid.persistence.core.rx.RxTaskScheduler;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
import org.apache.usergrid.persistence.core.util.StringUtils;
+import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.map.MapManager;
import org.apache.usergrid.persistence.map.MapManagerFactory;
import org.apache.usergrid.persistence.map.MapScope;
import org.apache.usergrid.persistence.map.impl.MapScopeImpl;
import org.apache.usergrid.persistence.model.util.UUIDGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import rx.Observable;
-import rx.observables.ConnectableObservable;
+import rx.schedulers.Schedulers;
@Singleton
@@ -51,14 +54,18 @@ public class ReIndexServiceImpl implements ReIndexService {
private static final MapScope RESUME_MAP_SCOPE =
new MapScopeImpl( CpNamingUtils.getManagementApplicationId(), "reindexresume" );
- //Keep cursors to resume re-index for 1 day. This is far beyond it's useful real world implications anyway.
+ //Keep cursors to resume re-index for 10 days. This is far beyond it's useful real world implications anyway.
private static final int INDEX_TTL = 60 * 60 * 24 * 10;
+ private static final String MAP_CURSOR_KEY = "cursor";
+ private static final String MAP_COUNT_KEY = "count";
+ private static final String MAP_STATUS_KEY = "status";
+ private static final String MAP_UPDATED_KEY = "lastUpdated";
+
private final AllApplicationsObservable allApplicationsObservable;
private final AllEntityIdsObservable allEntityIdsObservable;
private final IndexProcessorFig indexProcessorFig;
- private final RxTaskScheduler rxTaskScheduler;
private final MapManager mapManager;
private final AsyncEventService indexService;
@@ -66,69 +73,61 @@ public class ReIndexServiceImpl implements ReIndexService {
@Inject
public ReIndexServiceImpl( final AllEntityIdsObservable allEntityIdsObservable,
final MapManagerFactory mapManagerFactory,
- final AllApplicationsObservable allApplicationsObservable, final IndexProcessorFig indexProcessorFig,
- final RxTaskScheduler rxTaskScheduler, final AsyncEventService indexService ) {
+ final AllApplicationsObservable allApplicationsObservable,
+ final IndexProcessorFig indexProcessorFig, final AsyncEventService indexService ) {
this.allEntityIdsObservable = allEntityIdsObservable;
this.allApplicationsObservable = allApplicationsObservable;
this.indexProcessorFig = indexProcessorFig;
- this.rxTaskScheduler = rxTaskScheduler;
this.indexService = indexService;
this.mapManager = mapManagerFactory.createMapManager( RESUME_MAP_SCOPE );
}
-
-
-
@Override
public IndexResponse rebuildIndex( final IndexServiceRequestBuilder indexServiceRequestBuilder ) {
- //load our last emitted Scope if a cursor is present
- if ( indexServiceRequestBuilder.getCursor().isPresent() ) {
- throw new UnsupportedOperationException( "Build this" );
- }
+ //load our last emitted Scope if a cursor is present
+ final Optional<EdgeScope> cursor = parseCursor( indexServiceRequestBuilder.getCursor() );
+
+
+ final CursorSeek<Edge> cursorSeek = getResumeEdge( cursor );
final Optional<ApplicationScope> appId = indexServiceRequestBuilder.getApplicationScope();
- final Observable<ApplicationScope> applicationScopes = appId.isPresent()? Observable.just( appId.get() ) : allApplicationsObservable.getData();
+ Preconditions.checkArgument( cursor.isPresent() && appId.isPresent(),
+ "You cannot specify an app id and a cursor. When resuming with cursor you must omit the appid" );
+
+ final Observable<ApplicationScope> applicationScopes = getApplications( cursor, appId );
- final String newCursor = StringUtils.sanitizeUUID( UUIDGenerator.newTimeUUID() );
+ final String jobId = StringUtils.sanitizeUUID( UUIDGenerator.newTimeUUID() );
final long modifiedSince = indexServiceRequestBuilder.getUpdateTimestamp().or( Long.MIN_VALUE );
//create an observable that loads each entity and indexes it, start it running with publish
- final ConnectableObservable<EdgeScope> runningReIndex =
- allEntityIdsObservable.getEdgesToEntities( applicationScopes,
- indexServiceRequestBuilder.getCollectionName() )
-
- //for each edge, create our scope and index on it
- .doOnNext( edge -> indexService.index(
- new EntityIndexOperation( edge.getApplicationScope(), edge.getEdge().getTargetNode(),
- modifiedSince ) ) ).publish();
+ final Observable<EdgeScope> runningReIndex = allEntityIdsObservable.getEdgesToEntities( applicationScopes,
+ indexServiceRequestBuilder.getCollectionName(), cursorSeek.getSeekValue() )
+ //for each edge, create our scope and index on it
+ .doOnNext( edge -> indexService.index(
+ new EntityIndexOperation( edge.getApplicationScope(), edge.getEdge().getTargetNode(),
+ modifiedSince ) ) );
//start our sampler and state persistence
//take a sample every sample interval to allow us to resume state with minimal loss
- runningReIndex.sample( indexProcessorFig.getReIndexSampleInterval(), TimeUnit.MILLISECONDS,
- rxTaskScheduler.getAsyncIOScheduler() )
- .doOnNext( edge -> {
-
-// final String serializedState = SerializableMapper.asString( edge );
-//
-// mapManager.putString( newCursor, serializedState, INDEX_TTL );
- } ).subscribe();
+ runningReIndex.buffer( indexProcessorFig.getUpdateInterval() )
+ //create our flushing collector and flush the edge scopes to it
+ .collect( () -> new FlushingCollector( jobId ),
+ ( ( flushingCollector, edgeScopes ) -> flushingCollector.flushBuffer( edgeScopes ) ) ).doOnNext( flushingCollector-> flushingCollector.complete() )
+ //subscribe on our I/O scheduler and run the task
+ .subscribeOn( Schedulers.io() ).subscribe();
- //start pushing to both
- runningReIndex.connect();
-
-
- return new IndexResponse( newCursor, runningReIndex );
+ return new IndexResponse( jobId, "Started", 0, 0 );
}
@@ -136,6 +135,155 @@ public class ReIndexServiceImpl implements ReIndexService {
public IndexServiceRequestBuilder getBuilder() {
return new IndexServiceRequestBuilderImpl();
}
+
+
+ @Override
+ public IndexResponse getStatus( final String jobId ) {
+ Preconditions.checkNotNull( jobId, "jobId must not be null" );
+ return getIndexResponse( jobId );
+ }
+
+
+ /**
+ * Simple collector that counts state, then flushed every time a buffer is provided. Writes final state when complete
+ */
+ private class FlushingCollector {
+
+ private final String jobId;
+ private long count;
+
+
+ private FlushingCollector( final String jobId ) {
+ this.jobId = jobId;
+ }
+
+
+ public void flushBuffer( final List<EdgeScope> buffer ) {
+ count += buffer.size();
+
+ //write our cursor state
+ if ( buffer.size() > 0 ) {
+ writeCursorState( jobId, buffer.get( buffer.size() - 1 ) );
+ }
+
+ writeStateMeta( jobId, "InProgress", count, System.currentTimeMillis() );
+ }
+
+ public void complete(){
+ writeStateMeta( jobId, "Complete", count, System.currentTimeMillis() );
+ }
+ }
+
+
+ /**
+ * Get the resume edge scope
+ *
+ * @param edgeScope The optional edge scope from the cursor
+ */
+ private CursorSeek<Edge> getResumeEdge( final Optional<EdgeScope> edgeScope ) {
+
+
+ if ( edgeScope.isPresent() ) {
+ return new CursorSeek<>( Optional.of( edgeScope.get().getEdge() ) );
+ }
+
+ return new CursorSeek<>( Optional.absent() );
+ }
+
+
+ /**
+ * Generate an observable for our appliation scope
+ */
+ private Observable<ApplicationScope> getApplications( final Optional<EdgeScope> cursor,
+ final Optional<ApplicationScope> appId ) {
+ //cursor is present use it and skip until we hit that app
+ if ( cursor.isPresent() ) {
+
+ final EdgeScope cursorValue = cursor.get();
+ //we have a cursor and an application scope that was used.
+ return allApplicationsObservable.getData().skipWhile(
+ applicationScope -> !cursorValue.getApplicationScope().equals( applicationScope ) );
+ }
+ //this is intentional. If
+ else if ( appId.isPresent() ) {
+ return Observable.just( appId.get() );
+ }
+
+ return allApplicationsObservable.getData();
+ }
+
+
+ /**
+ * Swap our cursor for an optional edgescope
+ */
+ private Optional<EdgeScope> parseCursor( final Optional<String> cursor ) {
+
+ if ( !cursor.isPresent() ) {
+ return Optional.absent();
+ }
+
+ //get our cursor
+ final String persistedCursor = mapManager.getString( cursor.get() );
+
+ if ( persistedCursor == null ) {
+ return Optional.absent();
+ }
+
+ final JsonNode node = CursorSerializerUtil.fromString( persistedCursor );
+
+ final EdgeScope edgeScope = EdgeScopeSerializer.INSTANCE.fromJsonNode( node, CursorSerializerUtil.getMapper() );
+
+ return Optional.of( edgeScope );
+ }
+
+
+ /**
+ * Write the cursor state to the map in cassandra
+ */
+ private void writeCursorState( final String jobId, final EdgeScope edge ) {
+
+ final JsonNode node = EdgeScopeSerializer.INSTANCE.toNode( CursorSerializerUtil.getMapper(), edge );
+
+ final String serializedState = CursorSerializerUtil.asString( node );
+
+ mapManager.putString( jobId + MAP_CURSOR_KEY, serializedState, INDEX_TTL );
+ }
+
+
+ /**
+ * Write our state meta data into cassandra so everyone can see it
+ * @param jobId
+ * @param status
+ * @param processedCount
+ * @param lastUpdated
+ */
+ private void writeStateMeta( final String jobId, final String status, final long processedCount,
+ final long lastUpdated ) {
+
+ mapManager.putString( jobId + MAP_STATUS_KEY, status );
+ mapManager.putLong( jobId + MAP_COUNT_KEY, processedCount );
+ mapManager.putLong( jobId + MAP_UPDATED_KEY, lastUpdated );
+ }
+
+
+ /**
+ * Get the index response from the jobId
+ * @param jobId
+ * @return
+ */
+ private IndexResponse getIndexResponse( final String jobId ) {
+
+ final String status = mapManager.getString( jobId+MAP_STATUS_KEY );
+
+ if(status == null){
+ throw new IllegalArgumentException( "Could not find a job with id " + jobId );
+ }
+
+ final long processedCount = mapManager.getLong( jobId + MAP_COUNT_KEY );
+ final long lastUpdated = mapManager.getLong( jobId + MAP_COUNT_KEY );
+
+ return new IndexResponse( jobId, status, processedCount, lastUpdated );
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/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 fea0364..7acdd00 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
@@ -20,10 +20,15 @@
package org.apache.usergrid.corepersistence.pipeline.cursor;
-import com.fasterxml.jackson.core.Base64Variant;
-import com.fasterxml.jackson.core.Base64Variants;
+import java.io.IOException;
+import java.util.Base64;
+
+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.fasterxml.jackson.dataformat.smile.SmileFactory;
+import com.google.common.base.Preconditions;
/**
@@ -35,9 +40,54 @@ public class CursorSerializerUtil {
private static final ObjectMapper MAPPER = new ObjectMapper( SMILE_FACTORY );
+ /**
+ * Aritrary number, just meant to keep us from having a DOS issue
+ */
+ private static final int MAX_SIZE = 1024;
+
public static ObjectMapper getMapper() {
return MAPPER;
}
+
+ /**
+ * Turn the json node in to a base64 encoded SMILE binary
+ */
+ public static String asString( final JsonNode node ) {
+ final byte[] output;
+ try {
+ output = MAPPER.writeValueAsBytes( node );
+ }
+ catch ( JsonProcessingException e ) {
+ throw new RuntimeException( "Unable to create output from json node " + node );
+ }
+
+ //generate a base64 url save string
+ final String value = Base64.getUrlEncoder().encodeToString( output );
+
+ return value;
+ }
+
+
+ /**
+ * Parse the base64 encoded binary string
+ */
+ public static JsonNode fromString( final String base64EncodedJson ) {
+
+ Preconditions.checkArgument( base64EncodedJson.length() <= MAX_SIZE,
+ "Your cursor must be less than " + MAX_SIZE + " chars in length" );
+
+ final byte[] data = Base64.getUrlDecoder().decode( base64EncodedJson );
+
+ JsonNode jsonNode;
+ try {
+ jsonNode = MAPPER.readTree( data );
+ }
+ catch ( IOException e ) {
+ throw new RuntimeException( "Unable to parse json node from string " + base64EncodedJson );
+ }
+
+ return jsonNode;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/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 870edbb..dc6ae71 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
@@ -37,10 +37,6 @@ import com.google.common.base.Preconditions;
*/
public class RequestCursor {
- /**
- * Aritrary number, just meant to keep us from having a DOS issue
- */
- private static final int MAX_SIZE = 1024;
private static final int MAX_CURSOR_COUNT = 100;
@@ -83,11 +79,8 @@ public class RequestCursor {
try {
- Preconditions.checkArgument( cursor.length() <= MAX_SIZE, "Your cursor must be less than " + MAX_SIZE + " chars in length");
-
- final byte[] data = Base64.getUrlDecoder().decode( cursor );
- JsonNode jsonNode = MAPPER.readTree( data );
+ JsonNode jsonNode = CursorSerializerUtil.fromString( cursor );
Preconditions
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/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 dbd8b88..dc4bf39 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
@@ -20,11 +20,8 @@
package org.apache.usergrid.corepersistence.pipeline.cursor;
-import java.util.Base64;
-
import org.apache.usergrid.corepersistence.pipeline.read.EdgePath;
-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;
@@ -53,8 +50,8 @@ public class ResponseCursor {
/**
- * Lazyily encoded deliberately. If the user doesn't care about a cursor and is using streams, we dont' want to take the
- * time to calculate it
+ * Lazyily encoded deliberately. If the user doesn't care about a cursor and is using streams, we dont' want to
+ * take the time to calculate it
*/
public Optional<String> encodeAsString() {
@@ -68,42 +65,34 @@ public class ResponseCursor {
return encodedValue;
}
+ //no edge path, short circuit
- try {
-
- //no edge path, short circuit
-
- final ObjectNode map = MAPPER.createObjectNode();
+ final ObjectNode map = MAPPER.createObjectNode();
- Optional<EdgePath> current = edgePath;
+ Optional<EdgePath> current = edgePath;
- //traverse each edge and add them to our json
- do {
+ //traverse each edge and add them to our json
+ do {
- final EdgePath edgePath = current.get();
- final Object cursorValue = edgePath.getCursorValue();
- final CursorSerializer serializer = edgePath.getSerializer();
- final int filterId = edgePath.getFilterId();
+ final EdgePath edgePath = current.get();
+ final Object cursorValue = edgePath.getCursorValue();
+ final CursorSerializer serializer = edgePath.getSerializer();
+ final int filterId = edgePath.getFilterId();
- final JsonNode serialized = serializer.toNode( MAPPER, cursorValue );
- map.put( String.valueOf( filterId ), serialized );
+ final JsonNode serialized = serializer.toNode( MAPPER, cursorValue );
+ map.put( String.valueOf( filterId ), serialized );
- current = current.get().getPrevious();
- }
- while ( current.isPresent() );
+ current = current.get().getPrevious();
+ }
+ while ( current.isPresent() );
- final byte[] output = MAPPER.writeValueAsBytes( map );
+ //generate a base64 url save string
+ final String value = CursorSerializerUtil.asString( map );
- //generate a base64 url save string
- final String value = Base64.getUrlEncoder().encodeToString( output );
+ encodedValue = Optional.of( value );
- encodedValue = Optional.of( value );
- }
- catch ( JsonProcessingException e ) {
- throw new CursorParseException( "Unable to serialize cursor", e );
- }
return encodedValue;
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractPathFilter.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractPathFilter.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractPathFilter.java
index c68dc4a..0f9ac9b 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractPathFilter.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/AbstractPathFilter.java
@@ -76,34 +76,4 @@ public abstract class AbstractPathFilter<T, R, C extends Serializable> extends A
* Return the class to be used when parsing the cursor
*/
protected abstract CursorSerializer<C> getCursorSerializer();
-
-
- /**
- * An internal class that holds a mutable state. When resuming, we only ever honor the seek value on the first call. Afterwards, we will seek from the beginning on newly emitted values.
- * Calling get will return the first value to seek, or absent if not specified. Subsequent calls will return absent. Callers should treat the results as seek values for each operation
- */
- protected static class CursorSeek<C> {
-
- private Optional<C> seek;
-
- private CursorSeek(final Optional<C> cursorValue){
- seek = cursorValue;
- }
-
-
- /**
- * Get the seek value to use when searching
- * @return
- */
- public Optional<C> getSeekValue(){
- final Optional<C> toReturn = seek;
-
- seek = Optional.absent();
-
- return toReturn;
- }
-
-
-
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CursorSeek.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CursorSeek.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CursorSeek.java
new file mode 100644
index 0000000..b803658
--- /dev/null
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/pipeline/read/CursorSeek.java
@@ -0,0 +1,53 @@
+/*
+ * 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 com.google.common.base.Optional;
+
+
+/**
+ * An internal class that holds a mutable state. When resuming, we only ever honor the seek value on the first call. Afterwards, we will seek from the beginning on newly emitted values.
+ * Calling get will return the first value to seek, or absent if not specified. Subsequent calls will return absent. Callers should treat the results as seek values for each operation
+ */
+public class CursorSeek<C> {
+
+ private Optional<C> seek;
+
+ public CursorSeek( final Optional<C> cursorValue ){
+ seek = cursorValue;
+ }
+
+
+ /**
+ * Get the seek value to use when searching
+ * @return
+ */
+ public Optional<C> getSeekValue(){
+ final Optional<C> toReturn = seek;
+
+ seek = Optional.absent();
+
+ return toReturn;
+ }
+
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservable.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservable.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservable.java
index aada240..9070609 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservable.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservable.java
@@ -24,6 +24,7 @@ import com.google.common.base.Optional;
import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.graph.Edge;
import rx.Observable;
@@ -44,8 +45,9 @@ public interface AllEntityIdsObservable {
* Get all edges that represent edges to entities in the system
* @param appScopes
* @param edgeType The edge type to use (if specified)
+ * @param lastEdge The edge to resume processing from
* @return
*/
- Observable<EdgeScope> getEdgesToEntities(final Observable<ApplicationScope> appScopes, final Optional<String> edgeType);
+ Observable<EdgeScope> getEdgesToEntities(final Observable<ApplicationScope> appScopes, final Optional<String> edgeType, final Optional<Edge> lastEdge);
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java
----------------------------------------------------------------------
diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java
index 6a95e7b..0420a32 100644
--- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java
+++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/rx/impl/AllEntityIdsObservableImpl.java
@@ -28,6 +28,7 @@ import com.google.inject.Singleton;
import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope;
import org.apache.usergrid.persistence.core.scope.ApplicationScope;
+import org.apache.usergrid.persistence.graph.Edge;
import org.apache.usergrid.persistence.graph.GraphManager;
import org.apache.usergrid.persistence.graph.GraphManagerFactory;
import org.apache.usergrid.persistence.graph.serialization.EdgesObservable;
@@ -81,12 +82,12 @@ public class AllEntityIdsObservableImpl implements AllEntityIdsObservable {
@Override
- public Observable<EdgeScope> getEdgesToEntities( final Observable<ApplicationScope> appScopes, final Optional<String> edgeType) {
+ public Observable<EdgeScope> getEdgesToEntities( final Observable<ApplicationScope> appScopes, final Optional<String> edgeType, final Optional<Edge> lastEdge) {
return appScopes.flatMap( applicationScope -> {
final GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope );
- return edgesObservable.edgesFromSourceDescending( gm, applicationScope.getApplication(), edgeType )
+ return edgesObservable.edgesFromSourceDescending( gm, applicationScope.getApplication(), edgeType, lastEdge )
.map( edge -> new EdgeScope(applicationScope, edge ));
} );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
----------------------------------------------------------------------
diff --git a/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java b/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
index a17c925..cb9919f 100644
--- a/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
+++ b/stack/core/src/test/java/org/apache/usergrid/persistence/PerformanceEntityRebuildIndexTest.java
@@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit;
import com.google.common.base.Optional;
import org.apache.commons.lang.RandomStringUtils;
+import org.apache.usergrid.corepersistence.index.IndexServiceRequestBuilder;
import org.apache.usergrid.corepersistence.index.ReIndexService;
import org.apache.usergrid.persistence.index.ApplicationEntityIndex;
import org.junit.After;
@@ -196,7 +197,9 @@ public class PerformanceEntityRebuildIndexTest extends AbstractCoreIT {
try {
- reIndexService.rebuildIndex( Optional.of( em.getApplicationId()), Optional.<String>of("catherders"), Optional.absent(), Optional.absent() );
+ final IndexServiceRequestBuilder builder = reIndexService.getBuilder().withApplicationId( em.getApplicationId() ).withCollection( "catherders" );
+
+ reIndexService.rebuildIndex(builder );
reporter.report();
registry.remove( meterName );
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgesObservable.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgesObservable.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgesObservable.java
index 964e13d..78a1d4b 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgesObservable.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/EdgesObservable.java
@@ -42,16 +42,6 @@ public interface EdgesObservable {
/**
- * Return an observable of all edges from a source node. Ordered ascending, from the startTimestamp if specified
- * @param gm
- * @param sourceNode
- * @param edgeType The edge type if specified. Otherwise all types will be used
- * @return
- */
- Observable<Edge> edgesFromSourceDescending( final GraphManager gm, final Id sourceNode,
- final Optional<String> edgeType );
-
- /**
* Get all edges from the source node with the target type
* @param gm
* @param sourceNode
@@ -67,4 +57,15 @@ public interface EdgesObservable {
* @return
*/
Observable<Edge> edgesToTarget(final GraphManager gm, final Id targetNode);
+
+ /**
+ * Return an observable of all edges from a source node. Ordered ascending, from the startTimestamp if specified
+ * @param gm
+ * @param sourceNode
+ * @param edgeType The edge type if specified. Otherwise all types will be used
+ * @param resume The edge to start seeking after. Otherwise starts at the most recent
+ * @return
+ */
+ Observable<Edge> edgesFromSourceDescending( final GraphManager gm, final Id sourceNode,
+ final Optional<String> edgeType, final Optional<Edge> resume );
}
http://git-wip-us.apache.org/repos/asf/incubator-usergrid/blob/20c9b350/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgesObservableImpl.java
----------------------------------------------------------------------
diff --git a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgesObservableImpl.java b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgesObservableImpl.java
index 7240798..18274ac 100644
--- a/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgesObservableImpl.java
+++ b/stack/corepersistence/graph/src/main/java/org/apache/usergrid/persistence/graph/serialization/impl/EdgesObservableImpl.java
@@ -72,7 +72,7 @@ public class EdgesObservableImpl implements EdgesObservable {
@Override
public Observable<Edge> edgesFromSourceDescending( final GraphManager gm, final Id sourceNode,
- final Optional<String> edgeTypeInput ) {
+ final Optional<String> edgeTypeInput, final Optional<Edge> resume ) {
@@ -86,7 +86,7 @@ public class EdgesObservableImpl implements EdgesObservable {
return gm.loadEdgesFromSource(
new SimpleSearchByEdgeType( sourceNode, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING,
- Optional.<Edge>absent() ) );
+ resume ) );
} );
}