Add readonly and watch attributes to CriteriaRepository

criteria/common/src/org/immutables/criteria/Criteria.java

@@ -55,7 +55,28 @@
   @Documented
   @Target(ElementType.TYPE)
   @Retention(RetentionPolicy.SOURCE)
-  @interface Repository {}
+  @interface Repository {
+
+    /**
+     * Wherever repository should exclude write operations like {@code insert}, {@code update},  {@code delete} etc.
+     *
+     * If set to {@code false} (default), repository will be generated with modify operations otherwise
+     * it will be readonly.
+     *
+     * @return {@code false} (default) to generate {@code insert} / {@code update} operations,
+     * {@code true} for readonly repository.
+     */
+    boolean readonly() default false;
+
+    /**
+     * Wherever repository supports pub/sub operation. Typically it is an unbounded flow of events
+     * matching particular criteria (eg. updates on a entity).
+     *
+     * @return {@code false} (default) to omit pub/sub code generation, {@code true} to generate
+     * watchable interfaces.
+     */
+    boolean watch() default false;
+  }
 
 
 }

criteria/common/src/org/immutables/criteria/Repository.java

@@ -27,8 +27,6 @@ public interface Repository<T> {
 
   /**
    * Allows to chain operations (like adding {@code offset} / {@code limit}) on some particular query.
-   *
-   * TODO: Think about Reader vs Finder which also has delete methods
    */
   interface Reader<T> {
     Reader<T> limit(long limit);
@@ -46,6 +44,20 @@ interface Readable<T> extends Repository<T> {
 
   }
 
+  /**
+   * Marker for a successful operation
+   */
+  enum Success {
+    SUCCESS
+  }
+
+  interface Writable<T> extends Repository<T> {
+    Publisher<Success> insert(Iterable<? extends T> docs);
+
+    Publisher<Success> delete(DocumentCriteria<T> criteria);
+
+  }
+
   interface Watcher<T> {
     Publisher<T> watch();
   }
@@ -57,6 +69,4 @@ interface Watchable<T> extends Repository<T> {
     Watcher<T> watcher(DocumentCriteria<T> criteria);
   }
 
-  // TODO think about Updater / Replacer interfaces
-
 }

criteria/common/src/org/immutables/criteria/adapter/InternalReader.java

@@ -25,7 +25,7 @@
 /**
  * Simple implementation of reader interface. Keeps immutable query internally.
  */
-public class InternalReader<T> implements Repository.Reader<T> {
+public final class InternalReader<T> implements Repository.Reader<T> {
 
   private final ImmutableQuery<T> query;
   private final Backend backend;

criteria/common/src/org/immutables/criteria/adapter/Query.java → criteria/common/src/org/immutables/criteria/adapter/InternalWatcher.java

@@ -17,25 +17,24 @@
 package org.immutables.criteria.adapter;
 
 import org.immutables.criteria.DocumentCriteria;
-import org.immutables.value.Value;
+import org.immutables.criteria.Repository;
+import org.reactivestreams.Publisher;
 
-import java.util.OptionalLong;
+import java.util.Objects;
 
-/**
- * Query sent to a backend
- */
-@Value.Immutable
-public interface Query<T> extends Backend.Operation<T> {
-
-  @Value.Parameter
-  DocumentCriteria<?> criteria();
-
-  @Value.Parameter
-  Class<T> returnType();
+public final class InternalWatcher<T> implements Repository.Watcher<T> {
 
-  OptionalLong limit();
+  private final Backend backend;
+  private final DocumentCriteria<?> criteria;
 
-  OptionalLong offset();
+  public InternalWatcher(DocumentCriteria<?> criteria, Backend backend) {
+    this.backend = Objects.requireNonNull(backend, "backend");
+    this.criteria = Objects.requireNonNull(criteria, "criteria");
+  }
 
+  @Override
+  public Publisher<T> watch() {
+    return backend.execute(ImmutableWatch.of(criteria));
+  }
 
 }

criteria/common/src/org/immutables/criteria/adapter/Operations.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2019 Immutables Authors and Contributors
+ *
+ * Licensed 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.immutables.criteria.adapter;
+
+import org.immutables.criteria.DocumentCriteria;
+import org.immutables.criteria.Repository;
+import org.immutables.value.Value;
+
+import java.util.List;
+import java.util.OptionalLong;
+
+/**
+ * List of default operations which can be executed on the backend
+ */
+public final class Operations {
+
+  private Operations() {}
+
+  /**
+   * Query sent to a backend
+   */
+  @Value.Immutable
+  public interface Query<T> extends Backend.Operation<T> {
+
+    @Value.Parameter
+    DocumentCriteria<?> criteria();
+
+    @Value.Parameter
+    Class<T> returnType();
+
+    OptionalLong limit();
+
+    OptionalLong offset();
+  }
+
+  /**
+   * Insert list of "documents"
+   */
+  @Value.Immutable
+  public interface Insert extends Backend.Operation<Repository.Success> {
+    @Value.Parameter
+    List<?> entities();
+  }
+
+  /**
+   * Delete documents using some criteria
+   */
+  @Value.Immutable
+  public interface Delete extends Backend.Operation<Repository.Success> {
+    @Value.Parameter
+    DocumentCriteria<?> criteria();
+
+  }
+
+  @Value.Immutable
+  public interface Watch<T> extends Backend.Operation<T> {
+    @Value.Parameter
+    DocumentCriteria<?> criteria();
+
+  }
+
+
+}

criteria/common/test/org/immutables/criteria/personmodel/Person.java

@@ -33,7 +33,7 @@
  */
 @Value.Immutable
 @Criteria
-@Criteria.Repository
+@Criteria.Repository(watch = true)
 @JsonSerialize(as = ImmutablePerson.class)
 @JsonDeserialize(as = ImmutablePerson.class)
 public interface Person {

criteria/elasticsearch/src/org/immutables/criteria/elasticsearch/ElasticBackend.java

@@ -26,9 +26,9 @@
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.RestClient;
 import org.immutables.criteria.Criterias;
+import org.immutables.criteria.adapter.Operations;
 import org.immutables.criteria.expression.Expression;
 import org.immutables.criteria.adapter.Backend;
-import org.immutables.criteria.adapter.Query;
 import org.immutables.criteria.adapter.Reactive;
 import org.reactivestreams.Publisher;
 
@@ -65,10 +65,10 @@ public ElasticBackend(RestClient restClient,
   public <T> Publisher<T> execute(Operation<T> query) {
     Objects.requireNonNull(query, "query");
 
-    return queryInternal((Query<T>) query);
+    return queryInternal((Operations.Query<T>) query);
   }
 
-  private <T> Publisher<T> queryInternal(Query<T> query) {
+  private <T> Publisher<T> queryInternal(Operations.Query<T> query) {
     final Request request = new Request("POST", String.format("/%s/_search", index));
     final Expression expression = Criterias.toExpression(query.criteria());
     final ObjectNode json = Elasticsearch.converter(mapper).convert(expression);

criteria/mongo/src/org/immutables/criteria/mongo/MongoBackend.java

@@ -16,13 +16,20 @@
 
 package org.immutables.criteria.mongo;
 
+import com.mongodb.client.model.changestream.FullDocument;
 import com.mongodb.reactivestreams.client.MongoCollection;
+import org.bson.Document;
 import org.bson.conversions.Bson;
 import org.immutables.criteria.Criterias;
+import org.immutables.criteria.DocumentCriteria;
+import org.immutables.criteria.Repository;
 import org.immutables.criteria.adapter.Backend;
-import org.immutables.criteria.adapter.Query;
+import org.immutables.criteria.adapter.Operations;
+import org.immutables.criteria.adapter.Reactive;
+import org.immutables.criteria.expression.ExpressionConverter;
 import org.reactivestreams.Publisher;
 
+import java.util.Collections;
 import java.util.Objects;
 
 /**
@@ -33,17 +40,55 @@
 class MongoBackend implements Backend {
 
   private final MongoCollection<?> collection;
+  private final ExpressionConverter<Bson> converter;
 
   MongoBackend(MongoCollection<?> collection) {
     this.collection = Objects.requireNonNull(collection, "collection");
+    this.converter = Mongos.converter(collection.getCodecRegistry());
+  }
+
+  private Bson toBson(DocumentCriteria<?> criteria) {
+    return converter.convert(Criterias.toExpression(criteria));
   }
 
   @Override
   public <T> Publisher<T> execute(Operation<T> operation) {
+    if (operation instanceof Operations.Query) {
+      return query((Operations.Query<T>) operation);
+    } else if (operation instanceof Operations.Insert) {
+      return (Publisher<T>) insert((Operations.Insert) operation);
+    } else if (operation instanceof Operations.Delete) {
+      return (Publisher<T>) delete((Operations.Delete) operation);
+    } else if (operation instanceof Operations.Watch) {
+      return watch((Operations.Watch<T>) operation);
+    }
+
+    return Reactive.error(new UnsupportedOperationException(String.format("Operation %s not supported", operation)));
+  }
+
+  private <T> Publisher<T> query(Operations.Query<T> query) {
     @SuppressWarnings("unchecked")
     final MongoCollection<T> collection = (MongoCollection<T>) this.collection;
-    final Bson filter = Mongos.converter(collection.getCodecRegistry()).convert(Criterias.toExpression(((Query) operation).criteria()));
-    return collection.find(filter);
+    return collection.find(toBson(query.criteria()));
+  }
+
+  private Publisher<Repository.Success> delete(Operations.Delete delete) {
+    final Bson filter = toBson(delete.criteria());
+    return Reactive.map(collection.deleteMany(filter), r -> Repository.Success.SUCCESS);
+  }
+
+  private Publisher<Repository.Success> insert(Operations.Insert insert) {
+    final MongoCollection<Object> collection = (MongoCollection<Object>) this.collection;
+    return Reactive.map(collection.insertMany(insert.entities()), r -> Repository.Success.SUCCESS);
+  }
+
+  private <T> Publisher<T> watch(Operations.Watch<T> operation) {
+    final MongoCollection<T> collection = (MongoCollection<T>) this.collection;
+    final Bson filter = new Document("fullDocument", toBson(operation.criteria()));
+    return collection.watch(Collections.singletonList(filter))
+            .fullDocument(FullDocument.UPDATE_LOOKUP)
+            .withDocumentClass(collection.getDocumentClass());
+
   }
 
 }

value-processor/src/org/immutables/value/processor/CriteriaRepository.generator

@@ -38,11 +38,16 @@ package [type.package];
 import [starImport];
 [/for]
 
+import com.google.common.collect.ImmutableList;
+
 import org.immutables.criteria.Repository;
 import org.immutables.criteria.DocumentCriteria;
-import org.immutables.criteria.adapter.Query;
+import org.immutables.criteria.adapter.ImmutableQuery;
+import org.immutables.criteria.adapter.ImmutableInsert;
+import org.immutables.criteria.adapter.ImmutableDelete;
 import org.immutables.criteria.adapter.Backend;
 import org.immutables.criteria.adapter.InternalReader;
+import org.immutables.criteria.adapter.InternalWatcher;
 
 import org.reactivestreams.Publisher;
 
@@ -55,7 +60,11 @@ import java.util.Objects;
 [if type allowsClasspathAnnotation 'javax.annotation.concurrent.ThreadSafe']
 @javax.annotation.concurrent.ThreadSafe
 [/if]
-[type.typeDocument.access] class [type.name]Repository implements Repository.Readable<[type.name]> {
+[type.typeDocument.access] class [type.name]Repository implements [output.trim]
+Repository.Readable<[type.name]>
+[if not type.criteriaRepository.readonly], Repository.Writable<[type.name]>[/if]
+[if type.criteriaRepository.watch], Repository.Watchable<[type.name]>[/if]
+ {[/output.trim]
 
   private static final Class<[type.name]> ENTITY_CLASS =  [type.name].class;
 
@@ -74,6 +83,26 @@ import java.util.Objects;
   public Repository.Reader<[type.name]> findAll() {
     return find([type.name]Criteria.create());
   }
+
+[if not type.criteriaRepository.readonly]
+  @Override
+  public Publisher<Success> insert(Iterable<? extends [type.name]> docs) {
+    return backend.execute(ImmutableInsert.of(ImmutableList.copyOf(docs)));
+  }
+
+  @Override
+  public Publisher<Success> delete(DocumentCriteria<[type.name]> criteria) {
+    return backend.execute(ImmutableDelete.of(criteria));
+  }
+[/if]
+
+[if type.criteriaRepository.watch]
+  @Override
+  public Repository.Watcher<[type.name]> watcher(DocumentCriteria<[type.name]> criteria) {
+    return new InternalWatcher<>(criteria, backend);
+  }
+[/if]
+
 }
 
 [/template]

value-processor/src/org/immutables/value/processor/meta/CriteriaMirrors.java

@@ -25,7 +25,12 @@ private CriteriaMirrors() {}
 
   @Mirror.Annotation("org.immutables.criteria.Criteria.Repository")
   // different name because of collision with @Mongo.Repository
-  public @interface CriteriaRepository {}
+  public @interface CriteriaRepository {
+
+    boolean readonly() default false;
+
+    boolean watch() default false;
+  }
 
   @Mirror.Annotation("org.immutables.criteria.Criteria.Id")
   // different name because of collision with @Mongo.Id