Synchronized context and implemented context-preserving job execution.

rapidoid-ctx/src/main/java/org/rapidoid/ctx/Ctx.java

@@ -36,60 +36,62 @@ public class Ctx {
 
 	Ctx() {}
 
-	public UserInfo user() {
+	public synchronized UserInfo user() {
 		return user;
 	}
 
-	public void setUser(UserInfo user) {
+	public synchronized void setUser(UserInfo user) {
 		this.user = user;
 	}
 
 	@SuppressWarnings("unchecked")
-	public <T> T exchange() {
+	public synchronized <T> T exchange() {
 		return (T) exchange;
 	}
 
-	public void setExchange(Object exchange) {
+	public synchronized void setExchange(Object exchange) {
 		this.exchange = exchange;
 	}
 
-	public Classes classes() {
+	public synchronized Classes classes() {
 		return classes;
 	}
 
-	public void setClasses(Classes classes) {
+	public synchronized void setClasses(Classes classes) {
 		this.classes = classes;
 	}
 
 	@SuppressWarnings("unchecked")
-	public <P> P persistor() {
+	public synchronized <P> P persistor() {
 		if (this.persistor == null) {
 			this.persistor = Ctxs.createPersistor();
 		}
 
 		return (P) this.persistor;
 	}
 
-	public void setPersistor(Object persistor) {
+	public synchronized void setPersistor(Object persistor) {
 		this.persistor = persistor;
 	}
 
-	public void clear() {
+	public synchronized void clear() {
 		setClasses(null);
 		setExchange(null);
 		setUser(null);
 		setPersistor(null);
 	}
 
-	public Ctx copy() {
+	public synchronized Ctx copy() {
 		Ctx ctx = new Ctx();
+		this.assignTo(ctx);
+		return ctx;
+	}
 
+	public synchronized void assignTo(Ctx ctx) {
 		ctx.classes = this.classes;
 		ctx.exchange = this.exchange;
 		ctx.user = this.user;
 		ctx.persistor = this.persistor;
-
-		return ctx;
 	}
 
 }

rapidoid-ctx/src/main/java/org/rapidoid/ctx/Ctxs.java

@@ -32,8 +32,12 @@ public class Ctxs {
 
 	private Ctxs() {}
 
+	public static Ctx get() {
+		return CTXS.get();
+	}
+
 	public static Ctx ctx() {
-		Ctx ctx = CTXS.get();
+		Ctx ctx = get();
 
 		if (ctx == null) {
 			throw new IllegalStateException("App ctx wasn't set!");
@@ -43,26 +47,25 @@ public static Ctx ctx() {
 	}
 
 	public static boolean hasContext() {
-		return CTXS.get() != null;
+		return get() != null;
 	}
 
-	private static Ctx provide() {
-		Ctx ctx = CTXS.get();
+	public static void reset() {
+		CTXS.remove();
+	}
 
-		if (ctx == null) {
-			ctx = new Ctx();
+	public static void attach(Ctx ctx) {
+		if (!hasContext()) {
 			CTXS.set(ctx);
+		} else {
+			throw new IllegalStateException("The context was already opened!");
 		}
-
-		return ctx;
-	}
-
-	public static void reset() {
-		CTXS.remove();
 	}
 
-	public static void open() {
-		provide().clear();
+	public static Ctx open() {
+		Ctx ctx = new Ctx();
+		attach(ctx);
+		return ctx;
 	}
 
 	public static void close() {

rapidoid-utils/src/main/java/org/rapidoid/util/WrapperJob.java → rapidoid-utils/src/main/java/org/rapidoid/util/ContextPreservingJob.java

@@ -20,20 +20,38 @@
  * #L%
  */
 
+import org.rapidoid.annotation.Authors;
+import org.rapidoid.annotation.Since;
+import org.rapidoid.ctx.Ctx;
 import org.rapidoid.ctx.Ctxs;
+import org.rapidoid.log.Log;
 
-public class WrapperJob implements Runnable {
+@Authors("Nikolche Mihajlovski")
+@Since("4.1.0")
+public class ContextPreservingJob implements Runnable {
 
 	private final Runnable job;
 
-	public WrapperJob(Runnable job) {
+	private final Ctx ctx;
+
+	public ContextPreservingJob(Runnable job, Ctx ctx) {
 		this.job = job;
+		this.ctx = ctx;
 	}
 
 	@Override
 	public void run() {
+		if (ctx != null) {
+			Ctxs.attach(ctx);
+		} else {
+			Ctxs.open();
+		}
+
 		try {
 			job.run();
+		} catch (Throwable e) {
+			Log.error("Job execution failed!", e);
+			throw U.rte("Job execution failed!", e);
 		} finally {
 			Ctxs.close();
 		}

rapidoid-utils/src/main/java/org/rapidoid/util/Jobs.java

@@ -27,6 +27,7 @@
 import org.rapidoid.annotation.Authors;
 import org.rapidoid.annotation.Since;
 import org.rapidoid.config.Conf;
+import org.rapidoid.ctx.Ctxs;
 
 @Authors("Nikolche Mihajlovski")
 @Since("4.1.0")
@@ -54,7 +55,7 @@ public static ScheduledFuture<?> execute(Runnable job) {
 	}
 
 	public static Runnable wrap(Runnable job) {
-		return new WrapperJob(job);
+		return new ContextPreservingJob(job, Ctxs.get());
 	}
 
 }

rapidoid-utils/src/main/java/org/rapidoid/util/Schedule.java

@@ -54,7 +54,7 @@ public static synchronized ScheduledFuture<?> job(Runnable job, long delay) {
 	}
 
 	public static Runnable wrap(Runnable job) {
-		return new WrapperJob(job);
+		return Jobs.wrap(job);
 	}
 
 }

rapidoid-utils/src/test/java/org/rapidoid/util/JobsTest.java

@@ -0,0 +1,81 @@
+package org.rapidoid.util;
+
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Test;
+import org.rapidoid.annotation.Authors;
+import org.rapidoid.annotation.Since;
+import org.rapidoid.ctx.Ctxs;
+import org.rapidoid.ctx.UserInfo;
+import org.rapidoid.test.TestCommons;
+
+/*
+ * #%L
+ * rapidoid-utils
+ * %%
+ * Copyright (C) 2014 - 2015 Nikolche Mihajlovski 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.
+ * #L%
+ */
+
+@Authors("Nikolche Mihajlovski")
+@Since("4.1.0")
+public class JobsTest extends TestCommons {
+
+	@Test(timeout = 10000)
+	public void testJobsExecution() {
+
+		int total = 300000;
+		final AtomicInteger counter = new AtomicInteger();
+
+		multiThreaded(1000, total, new Runnable() {
+
+			@Override
+			public void run() {
+				Ctxs.open();
+
+				final UserInfo user = new UserInfo();
+				user.username = rndStr(50);
+				Ctxs.ctx().setUser(user);
+				ensureUser(user);
+
+				ScheduledFuture<?> future = Jobs.execute(new Runnable() {
+					@Override
+					public void run() {
+						ensureUser(user);
+						counter.incrementAndGet();
+					}
+				});
+
+				try {
+					future.get();
+				} catch (Exception e) {
+					e.printStackTrace();
+					fail("The job throwed an exception!");
+				}
+
+				Ctxs.close();
+			}
+
+		});
+
+		eq(counter.get(), total);
+	}
+
+	private void ensureUser(UserInfo user) {
+		eq(Ctxs.ctx().user(), user);
+	}
+
+}