2 files changed, 48 insertions, 14 deletions

diff --git a/common/pipeline.cpp b/common/pipeline.cpp
index 049d228..2c08aa0 100644
--- a/common/pipeline.cpp
+++ b/common/pipeline.cpp
@@ -181,16 +181,6 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
         key = Sink::Storage::generateUid();
     }
     Q_ASSERT(!key.isEmpty());
-    const qint64 newRevision = Storage::maxRevision(d->transaction) + 1;
-
-    // Add metadata buffer
-    flatbuffers::FlatBufferBuilder metadataFbb;
-    auto metadataBuilder = MetadataBuilder(metadataFbb);
-    metadataBuilder.add_revision(newRevision);
-    metadataBuilder.add_operation(Operation_Creation);
-    metadataBuilder.add_replayToSource(replayToSource);
-    auto metadataBuffer = metadataBuilder.Finish();
-    FinishMetadataBuffer(metadataFbb, metadataBuffer);
 
     auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(d->resourceType, bufferType);
     if (!adaptorFactory) {
@@ -201,8 +191,22 @@ KAsync::Job<qint64> Pipeline::newEntity(void const *command, size_t size)
     auto adaptor = adaptorFactory->createAdaptor(*entity);
     auto memoryAdaptor = QSharedPointer<Sink::ApplicationDomain::MemoryBufferAdaptor>::create(*(adaptor), adaptor->availableProperties());
     for (auto processor : d->processors[bufferType]) {
-        processor->newEntity(key, newRevision, *memoryAdaptor, d->transaction);
+        processor->resourceType = d->resourceType;
+        processor->pipeline = this;
+        processor->newEntity(key, Storage::maxRevision(d->transaction) + 1, *memoryAdaptor, d->transaction);
     }
+    //The maxRevision may have changed meanwhile if the entity created sub-entities
+    const qint64 newRevision = Storage::maxRevision(d->transaction) + 1;
+
+    // Add metadata buffer
+    flatbuffers::FlatBufferBuilder metadataFbb;
+    auto metadataBuilder = MetadataBuilder(metadataFbb);
+    metadataBuilder.add_revision(newRevision);
+    metadataBuilder.add_operation(Operation_Creation);
+    metadataBuilder.add_replayToSource(replayToSource);
+    auto metadataBuffer = metadataBuilder.Finish();
+    FinishMetadataBuffer(metadataFbb, metadataBuffer);
+
     flatbuffers::FlatBufferBuilder fbb;
     adaptorFactory->createBuffer(memoryAdaptor, fbb, metadataFbb.GetBufferPointer(), metadataFbb.GetSize());
 
@@ -461,6 +465,25 @@ void Preprocessor::finalize()
 {
 }
 
+void Preprocessor::createEntity(const Sink::ApplicationDomain::ApplicationDomainType &entity, const QByteArray &typeName)
+{
+    flatbuffers::FlatBufferBuilder entityFbb;
+    auto adaptorFactory = Sink::AdaptorFactoryRegistry::instance().getFactory(resourceType, typeName);
+    adaptorFactory->createBuffer(entity, entityFbb);
+    const auto entityBuffer = BufferUtils::extractBuffer(entityFbb);
+
+    flatbuffers::FlatBufferBuilder fbb;
+    auto entityId = fbb.CreateString(entity.identifier());
+    // This is the resource buffer type and not the domain type
+    auto type = fbb.CreateString(typeName);
+    auto delta = Sink::EntityBuffer::appendAsVector(fbb, entityBuffer.constData(), entityBuffer.size());
+    auto location = Sink::Commands::CreateCreateEntity(fbb, entityId, type, delta);
+    Sink::Commands::FinishCreateEntityBuffer(fbb, location);
+
+    const auto data = BufferUtils::extractBuffer(fbb);
+    pipeline->newEntity(data, data.size()).exec();
+}
+
 } // namespace Sink
 
 #pragma clang diagnostic push
diff --git a/common/pipeline.h b/common/pipeline.h
index 2ca87a4..c3abcf6 100644
--- a/common/pipeline.h
+++ b/common/pipeline.h
@@ -82,15 +82,26 @@ public:
     virtual ~Preprocessor();
 
     virtual void startBatch();
-    virtual void newEntity(const QByteArray &key, qint64 revision, Sink::ApplicationDomain::BufferAdaptor &newEntity, Sink::Storage::Transaction &transaction) = 0;
+    virtual void newEntity(const QByteArray &key, qint64 revision, Sink::ApplicationDomain::BufferAdaptor &newEntity, Sink::Storage::Transaction &transaction) {};
     virtual void modifiedEntity(const QByteArray &key, qint64 revision, const Sink::ApplicationDomain::BufferAdaptor &oldEntity,
-        Sink::ApplicationDomain::BufferAdaptor &newEntity, Sink::Storage::Transaction &transaction) = 0;
-    virtual void deletedEntity(const QByteArray &key, qint64 revision, const Sink::ApplicationDomain::BufferAdaptor &oldEntity, Sink::Storage::Transaction &transaction) = 0;
+        Sink::ApplicationDomain::BufferAdaptor &newEntity, Sink::Storage::Transaction &transaction) {};
+    virtual void deletedEntity(const QByteArray &key, qint64 revision, const Sink::ApplicationDomain::BufferAdaptor &oldEntity, Sink::Storage::Transaction &transaction) {};
     virtual void finalize();
 
+protected:
+    template <typename DomainType>
+    void createEntity(const DomainType &entity)
+    {
+        createEntity(entity, Sink::ApplicationDomain::getTypeName<DomainType>());
+    }
+    void createEntity(const Sink::ApplicationDomain::ApplicationDomainType &entity, const QByteArray &type);
+
 private:
+    friend class Pipeline;
     class Private;
     Private *const d;
+    Pipeline *pipeline;
+    QByteArray resourceType;
 };
 
 } // namespace Sink