JAVA-1458: Check thread in mapper sync methods

Alexandre Dutra · Alexandre Dutra · commit 325f13a44ca3 · 2017-06-02T14:29:27.000+02:00
This commit has substantial contributions by Greg Bestland (@GregBestland).
diff --git a/changelog/README.md b/changelog/README.md
@@ -14,6 +14,7 @@
 - [documentation] JAVA-1445: Clarify how nodes are penalized in LatencyAwarePolicy docs.
 - [improvement] JAVA-1446: Support 'DEFAULT UNSET' in Query Builder JSON Insert.
 - [improvement] JAVA-1443: Add groupBy method to Select statement.
+- [improvement] JAVA-1458: Check thread in mapper sync methods.
 
 
 ### 3.2.0
diff --git a/driver-mapping/src/main/java/com/datastax/driver/mapping/Mapper.java b/driver-mapping/src/main/java/com/datastax/driver/mapping/Mapper.java
@@ -171,15 +171,18 @@ public MappingManager getManager() {
      * Creates a query that can be used to save the provided entity.
      * <p/>
      * This method is useful if you want to setup a number of options (tracing,
-     * conistency level, ...) of the returned statement before executing it manually
+     * consistency level, ...) of the returned statement before executing it manually
      * or need access to the {@code ResultSet} object after execution (to get the
      * trace, the execution info, ...), but in other cases, calling {@link #save}
      * or {@link #saveAsync} is shorter.
+     * <p/>
+     * Note: this method might block if the query is not prepared yet.
      *
      * @param entity the entity to save.
-     * @return a query that saves {@code entity} (based on it's defined mapping).
+     * @return a query that saves {@code entity} (based on its defined mapping).
      */
     public Statement saveQuery(T entity) {
+        checkNotInEventLoop();
         try {
             return Uninterruptibles.getUninterruptibly(saveQueryAsync(entity, this.defaultSaveOptions));
         } catch (ExecutionException e) {
@@ -191,7 +194,7 @@ public Statement saveQuery(T entity) {
      * Creates a query that can be used to save the provided entity.
      * <p/>
      * This method is useful if you want to setup a number of options (tracing,
-     * conistency level, ...) of the returned statement before executing it manually
+     * consistency level, ...) of the returned statement before executing it manually
      * or need access to the {@code ResultSet} object after execution (to get the
      * trace, the execution info, ...), but in other cases, calling {@link #save}
      * or {@link #saveAsync} is shorter.
@@ -203,11 +206,13 @@ public Statement saveQuery(T entity) {
      * <li>Consistency level</li>
      * <li>Tracing</li>
      * </ul>
+     * Note: this method might block if the query is not prepared yet.
      *
      * @param entity the entity to save.
-     * @return a query that saves {@code entity} (based on it's defined mapping).
+     * @return a query that saves {@code entity} (based on its defined mapping).
      */
     public Statement saveQuery(T entity, Option... options) {
+        checkNotInEventLoop();
         try {
             return Uninterruptibles.getUninterruptibly(saveQueryAsync(entity, toMapWithDefaults(options, this.defaultSaveOptions)));
         } catch (ExecutionException e) {
@@ -271,10 +276,13 @@ else if (customCodec != null)
      * Saves an entity mapped by this mapper.
      * <p/>
      * This method is basically equivalent to: {@code getManager().getSession().execute(saveQuery(entity))}.
+     * <p/>
+     * Note: this method will block until the entity is fully saved.
      *
      * @param entity the entity to save.
      */
     public void save(T entity) {
+        checkNotInEventLoop();
         try {
             Uninterruptibles.getUninterruptibly(saveAsync(entity));
         } catch (ExecutionException e) {
@@ -292,11 +300,13 @@ public void save(T entity) {
      * <li>Consistency level</li>
      * <li>Tracing</li>
      * </ul>
+     * Note: this method will block until the entity is fully saved.
      *
      * @param entity  the entity to save.
      * @param options the options object specified defining special options when saving.
      */
     public void save(T entity, Option... options) {
+        checkNotInEventLoop();
         try {
             Uninterruptibles.getUninterruptibly(saveAsync(entity, options));
         } catch (ExecutionException e) {
@@ -346,7 +356,7 @@ public ListenableFuture<ResultSet> apply(BoundStatement bs) throws Exception {
      * KEY (in the order of said primary key).
      * <p/>
      * This method is useful if you want to setup a number of options (tracing,
-     * conistency level, ...) of the returned statement before executing it manually,
+     * consistency level, ...) of the returned statement before executing it manually,
      * but in other cases, calling {@link #get} or {@link #getAsync} is shorter.
      * <p/>
      * This method allows you to provide a suite of {@link Option} to include in
@@ -355,6 +365,7 @@ public ListenableFuture<ResultSet> apply(BoundStatement bs) throws Exception {
      * <li>Consistency level</li>
      * <li>Tracing</li>
      * </ul>
+     * Note: this method might block if the query is not prepared yet.
      *
      * @param objects the primary key of the entity to fetch, or more precisely
      *                the values for the columns of said primary key in the order of the primary key.
@@ -365,6 +376,7 @@ public ListenableFuture<ResultSet> apply(BoundStatement bs) throws Exception {
      *                                  at least one of those values is {@code null}.
      */
     public Statement getQuery(Object... objects) {
+        checkNotInEventLoop();
         try {
             return Uninterruptibles.getUninterruptibly(getQueryAsync(objects));
         } catch (ExecutionException e) {
@@ -422,6 +434,8 @@ public BoundStatement apply(PreparedStatement input) {
      * Fetch an entity based on its primary key.
      * <p/>
      * This method is basically equivalent to: {@code map(getManager().getSession().execute(getQuery(objects))).one()}.
+     * <p/>
+     * Note: this method will block until the entity is fully fetched.
      *
      * @param objects the primary key of the entity to fetch, or more precisely
      *                the values for the columns of said primary key in the order of the primary key.
@@ -432,6 +446,7 @@ public BoundStatement apply(PreparedStatement input) {
      *                                  at least one of those values is {@code null}.
      */
     public T get(Object... objects) {
+        checkNotInEventLoop();
         try {
             return Uninterruptibles.getUninterruptibly(getAsync(objects));
         } catch (ExecutionException e) {
@@ -474,7 +489,7 @@ public ListenableFuture<ResultSet> apply(BoundStatement bs) throws Exception {
      * is supported for DELETE queries.
      * <p/>
      * This method is useful if you want to setup a number of options (tracing,
-     * conistency level, ...) of the returned statement before executing it manually
+     * consistency level, ...) of the returned statement before executing it manually
      * or need access to the {@code ResultSet} object after execution (to get the
      * trace, the execution info, ...), but in other cases, calling {@link #delete}
      * or {@link #deleteAsync} is shorter.
@@ -486,13 +501,15 @@ public ListenableFuture<ResultSet> apply(BoundStatement bs) throws Exception {
      * <li>Consistency level</li>
      * <li>Tracing</li>
      * </ul>
+     * Note: this method might block if the query is not prepared yet.
      *
      * @param entity  the entity to delete.
      * @param options the options to add to the DELETE query.
-     * @return a query that delete {@code entity} (based on it's defined mapping) with
+     * @return a query that delete {@code entity} (based on its defined mapping) with
      * provided USING options.
      */
     public Statement deleteQuery(T entity, Option... options) {
+        checkNotInEventLoop();
         try {
             return Uninterruptibles.getUninterruptibly(deleteQueryAsync(entity, toMapWithDefaults(options, defaultDeleteOptions)));
         } catch (ExecutionException e) {
@@ -507,15 +524,18 @@ public Statement deleteQuery(T entity, Option... options) {
      * provided entity and call {@link #deleteQuery(Object...)} with it.
      * <p/>
      * This method is useful if you want to setup a number of options (tracing,
-     * conistency level, ...) of the returned statement before executing it manually
+     * consistency level, ...) of the returned statement before executing it manually
      * or need access to the {@code ResultSet} object after execution (to get the
      * trace, the execution info, ...), but in other cases, calling {@link #delete}
      * or {@link #deleteAsync} is shorter.
+     * <p/>
+     * Note: this method might block if the query is not prepared yet.
      *
      * @param entity the entity to delete.
-     * @return a query that delete {@code entity} (based on it's defined mapping).
+     * @return a query that delete {@code entity} (based on its defined mapping).
      */
     public Statement deleteQuery(T entity) {
+        checkNotInEventLoop();
         try {
             return Uninterruptibles.getUninterruptibly(deleteQueryAsync(entity, defaultDeleteOptions));
         } catch (ExecutionException e) {
@@ -533,7 +553,7 @@ public Statement deleteQuery(T entity) {
      * is supported for DELETE queries.
      * <p/>
      * This method is useful if you want to setup a number of options (tracing,
-     * conistency level, ...) of the returned statement before executing it manually
+     * consistency level, ...) of the returned statement before executing it manually
      * or need access to the {@code ResultSet} object after execution (to get the
      * trace, the execution info, ...), but in other cases, calling {@link #delete}
      * or {@link #deleteAsync} is shorter.
@@ -544,6 +564,7 @@ public Statement deleteQuery(T entity) {
      * <li>Consistency level</li>
      * <li>Tracing</li>
      * </ul>
+     * Note: this method might block if the query is not prepared yet.
      *
      * @param objects the primary key of the entity to delete, or more precisely
      *                the values for the columns of said primary key in the order of the primary key.
@@ -555,6 +576,7 @@ public Statement deleteQuery(T entity) {
      *                                  at least one of those values is {@code null}.
      */
     public Statement deleteQuery(Object... objects) {
+        checkNotInEventLoop();
         try {
             return Uninterruptibles.getUninterruptibly(deleteQueryAsync(objects));
         } catch (ExecutionException e) {
@@ -622,10 +644,13 @@ public BoundStatement apply(PreparedStatement input) {
      * Deletes an entity mapped by this mapper.
      * <p/>
      * This method is basically equivalent to: {@code getManager().getSession().execute(deleteQuery(entity))}.
+     * <p/>
+     * Note: this method will block until the entity is fully deleted.
      *
      * @param entity the entity to delete.
      */
     public void delete(T entity) {
+        checkNotInEventLoop();
         try {
             Uninterruptibles.getUninterruptibly(deleteAsync(entity));
         } catch (ExecutionException e) {
@@ -637,11 +662,14 @@ public void delete(T entity) {
      * Deletes an entity mapped by this mapper using provided options.
      * <p/>
      * This method is basically equivalent to: {@code getManager().getSession().execute(deleteQuery(entity, options))}.
+     * <p/>
+     * Note: this method will block until the entity is fully deleted.
      *
      * @param entity  the entity to delete.
      * @param options the options to add to the DELETE query.
      */
     public void delete(T entity, Option... options) {
+        checkNotInEventLoop();
         try {
             Uninterruptibles.getUninterruptibly(deleteAsync(entity, options));
         } catch (ExecutionException e) {
@@ -678,6 +706,8 @@ public ListenableFuture<Void> deleteAsync(T entity, Option... options) {
      * Deletes an entity based on its primary key.
      * <p/>
      * This method is basically equivalent to: {@code getManager().getSession().execute(deleteQuery(objects))}.
+     * <p/>
+     * Note: this method will block until the entity is fully deleted.
      *
      * @param objects the primary key of the entity to delete, or more precisely
      *                the values for the columns of said primary key in the order
@@ -688,6 +718,7 @@ public ListenableFuture<Void> deleteAsync(T entity, Option... options) {
      *                                  at least one of those values is {@code null}.
      */
     public void delete(Object... objects) {
+        checkNotInEventLoop();
         try {
             Uninterruptibles.getUninterruptibly(deleteAsync(objects));
         } catch (ExecutionException e) {
@@ -844,6 +875,13 @@ private static EnumMap<Option.Type, Option> toMapWithDefaults(Option[] options,
         return result;
     }
 
+    private void checkNotInEventLoop() {
+        Session session = manager.getSession();
+        if (session instanceof AbstractSession) {
+            ((AbstractSession) session).checkNotInEventLoop();
+        }
+    }
+
     /**
      * An option for a mapper operation.
      * <p/>
diff --git a/driver-mapping/src/test/java/com/datastax/driver/mapping/MapperTest.java b/driver-mapping/src/test/java/com/datastax/driver/mapping/MapperTest.java
