Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Added: basic $created support without any integration tests

  • Loading branch information...
commit 4c8fd15dbd9a62b2310b1718019f4a5c4a75def5 1 parent bd283f8
@ysw ysw authored
Showing with 269 additions and 12 deletions.
  1. +1 −0  src/EventStore/EventStore.Projections.Core.Tests/EventStore.Projections.Core.Tests.csproj
  2. +9 −0 src/EventStore/EventStore.Projections.Core.Tests/Services/core_projection/FakeProjectionHandler.cs
  3. +79 −0 ...ore.Projections.Core.Tests/Services/core_projection/when_creating_a_new_partitiion_the_projection_should.cs
  4. +8 −0 src/EventStore/EventStore.Projections.Core.Tests/Services/projections_manager/FakeBiStateProjection.cs
  5. +8 −0 src/EventStore/EventStore.Projections.Core.Tests/Services/projections_manager/FakeForeachStreamProjection.cs
  6. +6 −0 ...ventStore/EventStore.Projections.Core.Tests/Services/projections_manager/FakeFromCatalogStreamProjection.cs
  7. +8 −0 src/EventStore/EventStore.Projections.Core.Tests/Services/projections_manager/FakeProjection.cs
  8. +6 −8 src/EventStore/EventStore.Projections.Core/Prelude/1Prelude.js
  9. +41 −0 src/EventStore/EventStore.Projections.Core/Prelude/Projections.js
  10. +10 −0 src/EventStore/EventStore.Projections.Core/Services/IProjectionStateHandler.cs
  11. +28 −4 src/EventStore/EventStore.Projections.Core/Services/Processing/EventProcessingProjectionProcessingPhase.cs
  12. +13 −0 src/EventStore/EventStore.Projections.Core/Services/v8/V8ProjectionStateHandler.cs
  13. +7 −0 src/EventStore/EventStore.Projections.Core/Standard/CategorizeEventsByStreamPath.cs
  14. +7 −0 src/EventStore/EventStore.Projections.Core/Standard/CategorizeStreamByPath.cs
  15. +7 −0 src/EventStore/EventStore.Projections.Core/Standard/IndexEventsByEventType.cs
  16. +7 −0 src/EventStore/EventStore.Projections.Core/Standard/IndexStreams.cs
  17. +6 −0 src/EventStore/EventStore.Projections.Core/Standard/StubHandler.cs
  18. +12 −0 src/EventStore/EventStore.Projections.Core/v8/QueryScript.cs
  19. +6 −0 src/EventStore/EventStore.Web/Users/IndexUsersProjectionHandler.cs
View
1  src/EventStore/EventStore.Projections.Core.Tests/EventStore.Projections.Core.Tests.csproj
@@ -198,6 +198,7 @@
<Compile Include="Services\core_projection\query_by_stream\when_receiving_a_committed_event_the_projection_with_partitioned_state_by_custom_rule_should.cs" />
<Compile Include="Services\core_projection\TestFixtureUtils.cs" />
<Compile Include="Services\core_projection\TestFixtureWithReadWriteDispatchers.cs" />
+ <Compile Include="Services\core_projection\when_creating_a_new_partitiion_the_projection_should.cs" />
<Compile Include="Services\core_projection\when_receiving_committed_events_the_projection_without_when.cs" />
<Compile Include="Services\core_projection\when_receiving_committed_event_the_projection_with_existing_partitioned_state_should.cs" />
<Compile Include="Services\core_projection\when_the_projection_with_pending_checkpoint_is_stopped.cs" />
View
9 src/EventStore/EventStore.Projections.Core.Tests/Services/core_projection/FakeProjectionHandler.cs
@@ -40,6 +40,7 @@ public class FakeProjectionStateHandler : IProjectionStateHandler
public int _initializeSharedCalled = 0;
public int _loadCalled = 0;
public int _eventsProcessed = 0;
+ public int _partitionCreatedProcessed = 0;
public string _loadedState = null;
public string _lastProcessedStreamId;
public string _lastProcessedEventType;
@@ -234,6 +235,14 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
}
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ _partitionCreatedProcessed++;
+ emittedEvents = null;
+ return true;
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
79 ....Projections.Core.Tests/Services/core_projection/when_creating_a_new_partitiion_the_projection_should.cs
@@ -0,0 +1,79 @@
+// Copyright (c) 2012, Event Store LLP
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+// Redistributions of source code must retain the above copyright notice,
+// this list of conditions and the following disclaimer.
+// Redistributions in binary form must reproduce the above copyright
+// notice, this list of conditions and the following disclaimer in the
+// documentation and/or other materials provided with the distribution.
+// Neither the name of the Event Store LLP nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//
+
+using System;
+using System.Linq;
+using EventStore.Common.Utils;
+using EventStore.Core.Data;
+using EventStore.Core.Messages;
+using EventStore.Projections.Core.Messages;
+using EventStore.Projections.Core.Services.Processing;
+using NUnit.Framework;
+using ResolvedEvent = EventStore.Projections.Core.Services.Processing.ResolvedEvent;
+
+namespace EventStore.Projections.Core.Tests.Services.core_projection
+{
+ [TestFixture]
+ public class when_creating_a_new_partitiion_the_projection_should : TestFixtureWithCoreProjectionStarted
+ {
+ private Guid _eventId;
+
+ protected override void Given()
+ {
+ _configureBuilderByQuerySource = source =>
+ {
+ source.FromAll();
+ source.AllEvents();
+ source.SetByStream();
+ source.SetDefinesStateTransform();
+ };
+ TicksAreHandledImmediately();
+ AllWritesSucceed();
+ NoOtherStreams();
+ }
+
+ protected override void When()
+ {
+ //projection subscribes here
+ _eventId = Guid.NewGuid();
+ _consumer.HandledMessages.Clear();
+ _bus.Publish(
+ EventReaderSubscriptionMessage.CommittedEventReceived.Sample(
+ new ResolvedEvent(
+ "account-01", -1, "account-01", -1, false, new TFPos(120, 110), _eventId, "handle_this_type",
+ false, "data", "metadata"), _subscriptionId, 0));
+ }
+
+ [Test]
+ public void passes_partition_created_notification_to_the_handler()
+ {
+ Assert.AreEqual(1, _stateHandler._partitionCreatedProcessed);
+ Assert.Inconclusive();
+ }
+ }
+}
View
8 src/EventStore/EventStore.Projections.Core.Tests/Services/projections_manager/FakeBiStateProjection.cs
@@ -102,6 +102,14 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ _logger("ProcessPartitionCreated");
+ emittedEvents = null;
+ return false;
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
8 ...EventStore/EventStore.Projections.Core.Tests/Services/projections_manager/FakeForeachStreamProjection.cs
@@ -105,6 +105,14 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ _logger("Process ProcessPartitionCreated");
+ emittedEvents = null;
+ return false;
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
6 ...tStore/EventStore.Projections.Core.Tests/Services/projections_manager/FakeFromCatalogStreamProjection.cs
@@ -104,6 +104,12 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ throw new NotImplementedException();
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
8 src/EventStore/EventStore.Projections.Core.Tests/Services/projections_manager/FakeProjection.cs
@@ -100,6 +100,14 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ _logger("Process ProcessPartitionCreated");
+ emittedEvents = null;
+ return false;
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
14 src/EventStore/EventStore.Projections.Core/Prelude/1Prelude.js
@@ -71,17 +71,15 @@ function scope($on, $notify) {
for (var name in handlers) {
if (name == 0 || name === "$init") {
eventProcessor.on_init_state(handlers[name]);
- }
- else if (name === "$initShared") {
+ } else if (name === "$initShared") {
eventProcessor.on_init_shared_state(handlers[name]);
- }
- else if (name === "$any") {
+ } else if (name === "$any") {
eventProcessor.on_any(handlers[name]);
- }
- else if (name === "$deleted") {
+ } else if (name === "$deleted") {
eventProcessor.on_deleted_notification(handlers[name]);
- }
- else {
+ } else if (name === "$created") {
+ eventProcessor.on_created_notification(handlers[name]);
+ } else {
eventProcessor.on_event(name, handlers[name]);
}
}
View
41 src/EventStore/EventStore.Projections.Core/Prelude/Projections.js
@@ -34,6 +34,7 @@ var $projections = {
var eventHandlers = {};
var anyEventHandlers = [];
var deletedNotificationHandlers = [];
+ var createdNotificationHandlers = [];
var rawEventHandlers = [];
var transformers = [];
var getStatePartitionHandler = function () {
@@ -142,6 +143,13 @@ var $projections = {
}
},
+ process_created_notification: function (event, isJson, streamId, eventType, category, sequenceNumber, metadata, linkMetadata, partition) {
+ processCreatedNotification(event, isJson, streamId, eventType, category, sequenceNumber, metadata, linkMetadata, partition);
+ var stateJson;
+ stateJson = JSON.stringify(projectionState);
+ return stateJson;
+ },
+
transform_state_to_result: function () {
var result = projectionState;
for (var i = 0; i < transformers.length; i++) {
@@ -213,6 +221,12 @@ var $projections = {
sources.options.definesFold = true;
}
+ function on_created_notification(eventHandler) {
+ createdNotificationHandlers.push(eventHandler);
+ sources.options.handlesCreatedNotifications = true;
+ sources.options.definesFold = true;
+ }
+
function on_raw(eventHandler) {
runDefaultHandler = false;
sources.allEvents = true;
@@ -388,6 +402,32 @@ var $projections = {
}
}
+ function processCreatedNotification(eventRaw, isJson, streamId, eventType, category, sequenceNumber, metadataRaw, linkMetadataRaw, partition, streamMetadataRaw) {
+
+ var eventHandler;
+ var state = !sources.options.biState ? projectionState : [projectionState, projectionSharedState];
+
+ var index;
+
+ var eventEnvelope = new envelope(null, eventRaw, eventType, streamId, sequenceNumber, metadataRaw, linkMetadataRaw, partition, streamMetadataRaw);
+
+ if (isJson) {
+ tryDeserializeBody(eventEnvelope);
+ }
+
+ for (index = 0; index < createdNotificationHandlers.length; index++) {
+ eventHandler = createdNotificationHandlers[index];
+ state = callHandler(eventHandler, state, eventEnvelope);
+ }
+
+ if (!sources.options.biState) {
+ projectionState = state;
+ } else {
+ projectionState = state[0];
+ projectionSharedState = state[1];
+ }
+ }
+
function fromStream(sourceStream) {
sources.streams.push(sourceStream);
}
@@ -460,6 +500,7 @@ var $projections = {
on_any: on_any,
on_raw: on_raw,
on_deleted_notification: on_deleted_notification,
+ on_created_notification: on_created_notification,
fromAll: fromAll,
fromCategory: fromCategory,
View
10 src/EventStore/EventStore.Projections.Core/Services/IProjectionStateHandler.cs
@@ -71,6 +71,16 @@ public interface IProjectionStateHandler : IDisposable, ISourceDefinitionSource
out string newSharedState, out EmittedEventEnvelope[] emittedEvents);
/// <summary>
+ /// Processes partition created notificatiion and updates internal state if necessary.
+ /// </summary>
+ /// <param name="partition"></param>
+ /// <param name="createPosition"></param>
+ /// <param name="emittedEvents"></param>
+ /// <returns>true - if notification was processed (new state must be returned)</returns>
+ bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents);
+
+ /// <summary>
/// Processes partition deleted notification and updates internal state if necessary.
/// </summary>
/// <returns>true - if event was processed (new state must be returned) </returns>
View
32 src/EventStore/EventStore.Projections.Core/Services/Processing/EventProcessingProjectionProcessingPhase.cs
@@ -27,6 +27,7 @@
//
using System;
using System.Diagnostics;
+using System.Linq;
using EventStore.Common.Log;
using EventStore.Core.Bus;
using EventStore.Projections.Core.Messages;
@@ -310,8 +311,15 @@ private string GetHandlerTypeName()
out string newSharedState, out string projectionResult, out EmittedEventEnvelope[] emittedEvents)
{
projectionResult = null;
- SetHandlerState(partition);
+ var newPatitionInitialized = InitOrLoadHandlerState(partition);
_stopwatch.Start();
+ EmittedEventEnvelope[] eventsEmittedOnInitialization = null;
+ if (newPatitionInitialized)
+ {
+ _projectionStateHandler.ProcessPartitionCreated(
+ partition, message.CheckpointTag, out eventsEmittedOnInitialization);
+ }
+
var result = _projectionStateHandler.ProcessEvent(
partition, message.CheckpointTag, message.EventCategory, message.Data, out newState, out newSharedState,
out emittedEvents);
@@ -336,6 +344,14 @@ private string GetHandlerTypeName()
}
}
_stopwatch.Stop();
+ if (eventsEmittedOnInitialization != null)
+ {
+ if (emittedEvents == null || emittedEvents.Length == 0)
+ emittedEvents = eventsEmittedOnInitialization;
+ else
+ emittedEvents = eventsEmittedOnInitialization.Concat(emittedEvents).ToArray();
+
+ }
return result;
}
@@ -344,7 +360,7 @@ private string GetHandlerTypeName()
out string projectionResult)
{
projectionResult = null;
- SetHandlerState(partition);
+ InitOrLoadHandlerState(partition);
_stopwatch.Start();
var result = _projectionStateHandler.ProcessPartitionDeleted(
partition, deletePosition, out newState);
@@ -380,17 +396,24 @@ private string TransformCatalogEventByHandler(EventReaderSubscriptionMessage.Com
return result;
}
- private void SetHandlerState(string partition)
+ /// <summary>
+ /// initializes or loads existing partition state
+ /// </summary>
+ /// <param name="partition"></param>
+ /// <returns>true - if new partition state was initialized</returns>
+ private bool InitOrLoadHandlerState(string partition)
{
if (_handlerPartition == partition)
- return;
+ return false;
var newState = _partitionStateCache.GetLockedPartitionState(partition);
_handlerPartition = partition;
+ var initialized = false;
if (newState != null && !String.IsNullOrEmpty(newState.State))
_projectionStateHandler.Load(newState.State);
else
{
+ initialized = true;
_projectionStateHandler.Initialize();
}
@@ -403,6 +426,7 @@ private void SetHandlerState(string partition)
else
_projectionStateHandler.InitializeShared();
}
+ return initialized;
}
public override void NewCheckpointStarted(CheckpointTag at)
View
13 src/EventStore/EventStore.Projections.Core/Services/v8/V8ProjectionStateHandler.cs
@@ -219,6 +219,19 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ CheckDisposed();
+ _eventPosition = createPosition;
+ _emittedEvents = null;
+ var newStates = _query.NotifyCreated(
+ "", // trimming data passed to a JS
+ new[] {partition, "" /* isSoftDedleted */});
+ emittedEvents = _emittedEvents == null ? null : _emittedEvents.ToArray();
+ return true;
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
CheckDisposed();
View
7 src/EventStore/EventStore.Projections.Core/Standard/CategorizeEventsByStreamPath.cs
@@ -115,6 +115,13 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ emittedEvents = null;
+ return false;
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
7 src/EventStore/EventStore.Projections.Core/Standard/CategorizeStreamByPath.cs
@@ -104,6 +104,13 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ emittedEvents = null;
+ return false;
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
7 src/EventStore/EventStore.Projections.Core/Standard/IndexEventsByEventType.cs
@@ -120,6 +120,13 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ emittedEvents = null;
+ return false;
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
7 src/EventStore/EventStore.Projections.Core/Standard/IndexStreams.cs
@@ -102,6 +102,13 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ emittedEvents = null;
+ return false;
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
6 src/EventStore/EventStore.Projections.Core/Standard/StubHandler.cs
@@ -88,6 +88,12 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ throw new NotImplementedException();
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
View
12 src/EventStore/EventStore.Projections.Core/v8/QueryScript.cs
@@ -45,6 +45,7 @@ public class QueryScript : IDisposable
private Func<string, string[], string> _transformCatalogEvent;
private Func<string, string[], Tuple<string, string>> _processEvent;
private Func<string, string[], string> _processDeletedNotification;
+ private Func<string, string[], string> _processCreatedNotification;
private Func<string> _transformStateToResult;
private Action<string> _setState;
private Action<string> _setSharedState;
@@ -135,6 +136,9 @@ private void CommandHandlerRegisteredCallback(string commandName, IntPtr handler
case "process_deleted_notification":
_processDeletedNotification = (json, other) => ExecuteHandler(handlerHandle, json, other);
break;
+ case "process_created_notification":
+ _processCreatedNotification = (json, other) => ExecuteHandler(handlerHandle, json, other);
+ break;
case "transform_catalog_event":
_transformCatalogEvent = (json, other) => ExecuteHandler(handlerHandle, json, other);
break;
@@ -280,6 +284,14 @@ public string NotifyDeleted(string json, string[] other)
return _processDeletedNotification(json, other);
}
+ public string NotifyCreated(string json, string[] other)
+ {
+ if (_processCreatedNotification == null)
+ throw new InvalidOperationException("'process_created_notification' command handler has not been registered");
+
+ return _processCreatedNotification(json, other);
+ }
+
public string TransformStateToResult()
{
if (_transformStateToResult == null)
View
6 src/EventStore/EventStore.Web/Users/IndexUsersProjectionHandler.cs
@@ -113,6 +113,12 @@ public string TransformCatalogEvent(CheckpointTag eventPosition, ResolvedEvent d
return true;
}
+ public bool ProcessPartitionCreated(
+ string partition, CheckpointTag createPosition, out EmittedEventEnvelope[] emittedEvents)
+ {
+ throw new NotImplementedException();
+ }
+
public bool ProcessPartitionDeleted(string partition, CheckpointTag deletePosition, out string newState)
{
throw new NotImplementedException();
Please sign in to comment.
Something went wrong with that request. Please try again.