deltaspike-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmannibu...@apache.org
Subject deltaspike git commit: DELTASPIKE-1094 DELTASPIKE-1093 @Throttled and @Futureable
Date Thu, 17 Mar 2016 11:09:38 GMT
Repository: deltaspike
Updated Branches:
  refs/heads/master a571ac650 -> dbd4fb78e


DELTASPIKE-1094 DELTASPIKE-1093 @Throttled and @Futureable


Project: http://git-wip-us.apache.org/repos/asf/deltaspike/repo
Commit: http://git-wip-us.apache.org/repos/asf/deltaspike/commit/dbd4fb78
Tree: http://git-wip-us.apache.org/repos/asf/deltaspike/tree/dbd4fb78
Diff: http://git-wip-us.apache.org/repos/asf/deltaspike/diff/dbd4fb78

Branch: refs/heads/master
Commit: dbd4fb78e1af1acb1e59e3f2de59979b012486b1
Parents: a571ac6
Author: Romain manni-Bucau <rmannibucau@gmail.com>
Authored: Thu Mar 17 12:09:18 2016 +0100
Committer: Romain manni-Bucau <rmannibucau@gmail.com>
Committed: Thu Mar 17 12:09:18 2016 +0100

----------------------------------------------------------------------
 .../deltaspike/core/api/future/Futureable.java  |  44 +++
 .../core/api/throttling/Throttled.java          |  56 ++++
 .../core/api/throttling/Throttling.java         |  73 +++++
 .../core/impl/future/FutureableInterceptor.java | 324 +++++++++++++++++++
 .../impl/throttling/ThrottledInterceptor.java   | 197 +++++++++++
 .../impl/src/main/resources/META-INF/beans.xml  |   4 +
 .../test/core/impl/future/FutureableTest.java   |  66 ++++
 .../test/core/impl/future/Service.java          |  61 ++++
 .../test/core/impl/throttling/Service.java      |  60 ++++
 .../test/core/impl/throttling/Service2.java     |  69 ++++
 .../core/impl/throttling/ThrottledTest.java     | 277 ++++++++++++++++
 11 files changed, 1231 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/future/Futureable.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/future/Futureable.java b/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/future/Futureable.java
new file mode 100644
index 0000000..2a725c2
--- /dev/null
+++ b/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/future/Futureable.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.deltaspike.core.api.future;
+
+import javax.enterprise.util.Nonbinding;
+import javax.interceptor.InterceptorBinding;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.TYPE;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+/**
+ * Mark the method as execute in a thread pool and not synchronously.
+ * Note: it should return a CompletionStage or Future.
+ */
+@InterceptorBinding
+@Retention(RUNTIME)
+@Target({ TYPE, METHOD })
+public @interface Futureable
+{
+    /**
+     * @return pool name, if not existing will use a default one based on processor count.
+     */
+    @Nonbinding
+    String value() default "";
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/throttling/Throttled.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/throttling/Throttled.java b/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/throttling/Throttled.java
new file mode 100644
index 0000000..4dcc0d2
--- /dev/null
+++ b/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/throttling/Throttled.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.deltaspike.core.api.throttling;
+
+import javax.enterprise.util.Nonbinding;
+import javax.interceptor.InterceptorBinding;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+import java.util.concurrent.TimeUnit;
+
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.TYPE;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+/**
+ * Mark a bean/method as relying on a throttler.
+ */
+@InterceptorBinding
+@Retention(RUNTIME)
+@Target({ TYPE, METHOD })
+public @interface Throttled
+{
+    /**
+     * @return the duration to wait to acquire the permits.
+     */
+    @Nonbinding
+    long timeout() default 0L;
+
+    /**
+     * @return the unit of timeout().
+     */
+    @Nonbinding
+    TimeUnit timeoutUnit() default TimeUnit.MILLISECONDS;
+
+    /**
+     * @return how many permits to require.
+     */
+    @Nonbinding
+    int weight() default 1;
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/throttling/Throttling.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/throttling/Throttling.java b/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/throttling/Throttling.java
new file mode 100644
index 0000000..fdf2cf1
--- /dev/null
+++ b/deltaspike/core/api/src/main/java/org/apache/deltaspike/core/api/throttling/Throttling.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.deltaspike.core.api.throttling;
+
+import javax.enterprise.util.Nonbinding;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+import javax.enterprise.inject.spi.AnnotatedMethod;
+import java.util.concurrent.Semaphore;
+
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.TYPE;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+/**
+ * Configure the throttler associated to the class/method.
+ */
+@Retention(RUNTIME)
+@Target({ TYPE, METHOD })
+public @interface Throttling
+{
+    /**
+     * @return how to get the semaphore. Default to a plain Semaphore of the JVM.
+     */
+    @Nonbinding
+    Class<? extends SemaphoreFactory> factory() default SemaphoreFactory.class;
+
+    /**
+     * @return true if the semaphore is fair false otherwise.
+     */
+    @Nonbinding
+    boolean fair() default false;
+
+    /**
+     * @return how many permits has the semaphore.
+     */
+    @Nonbinding
+    int permits() default 1;
+
+    /**
+     * @return name/bucket of this configuration (allow to have multiple buckets per class but default is 1 per class).
+     */
+    @Nonbinding
+    String name() default "";
+
+    interface SemaphoreFactory
+    {
+        /**
+         * @param method the intercepted method.
+         * @param name bucket name.
+         * @param fair should the semaphore be fair.
+         * @param permits maximum permits the semaphore shoulg get.
+         * @return the semaphore build accordingly the parameters.
+         */
+        Semaphore newSemaphore(AnnotatedMethod<?> method, String name, boolean fair, int permits);
+    }
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/future/FutureableInterceptor.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/future/FutureableInterceptor.java b/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/future/FutureableInterceptor.java
new file mode 100644
index 0000000..a1ae35b
--- /dev/null
+++ b/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/future/FutureableInterceptor.java
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.deltaspike.core.impl.future;
+
+import org.apache.deltaspike.core.api.config.ConfigResolver;
+import org.apache.deltaspike.core.api.future.Futureable;
+
+import javax.annotation.PreDestroy;
+import javax.enterprise.context.ApplicationScoped;
+import javax.enterprise.inject.spi.AnnotatedMethod;
+import javax.enterprise.inject.spi.AnnotatedType;
+import javax.enterprise.inject.spi.BeanManager;
+import javax.inject.Inject;
+import javax.interceptor.AroundInvoke;
+import javax.interceptor.Interceptor;
+import javax.interceptor.InvocationContext;
+import java.io.Serializable;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+@Interceptor
+@Futureable("")
+public class FutureableInterceptor implements Serializable
+{
+    private static final Class<?> COMPLETION_STAGE;
+    private static final Class<?> COMPLETABLE_FUTURE;
+    private static final Method COMPLETABLE_STAGE_TO_FUTURE;
+    private static final Method COMPLETABLE_FUTURE_COMPLETE;
+    private static final Method COMPLETABLE_FUTURE_COMPLETE_ERROR;
+
+    static
+    {
+        Class<?> completionStageClass = null;
+        Class<?> completableFutureClass = null;
+        Method completionStageClassToCompletableFuture = null;
+        Method completableFutureComplete = null;
+        Method completableFutureCompleteError = null;
+        try
+        {
+            final ClassLoader classLoader = ClassLoader.getSystemClassLoader();
+            completionStageClass = classLoader.loadClass("java.util.concurrent.CompletionStage");
+            completionStageClassToCompletableFuture = completionStageClass.getMethod("toCompletableFuture");
+            completableFutureClass = classLoader.loadClass("java.util.concurrent.CompletableFuture");
+            completableFutureComplete = completableFutureClass.getMethod("complete", Object.class);
+            completableFutureCompleteError = completableFutureClass.getMethod("completeExceptionally", Throwable.class);
+        }
+        catch (final Exception e)
+        {
+            // not on java 8
+        }
+        COMPLETION_STAGE = completionStageClass;
+        COMPLETABLE_FUTURE = completableFutureClass;
+        COMPLETABLE_STAGE_TO_FUTURE = completionStageClassToCompletableFuture;
+        COMPLETABLE_FUTURE_COMPLETE = completableFutureComplete;
+        COMPLETABLE_FUTURE_COMPLETE_ERROR = completableFutureCompleteError;
+    }
+
+    @Inject
+    private ThreadPoolManager manager;
+
+    @Inject
+    private BeanManager beanManager;
+
+    private transient ConcurrentMap<Method, ExecutorService> configByMethod =
+            new ConcurrentHashMap<Method, ExecutorService>();
+
+    @AroundInvoke
+    public Object invoke(final InvocationContext ic) throws Exception
+    {
+        // validate usage
+        final Class<?> returnType = ic.getMethod().getReturnType();
+        if (!COMPLETION_STAGE.isAssignableFrom(returnType) && !Future.class.isAssignableFrom(returnType))
+        {
+            throw new IllegalArgumentException("Return type should be a CompletableStage or Future");
+        }
+
+        if (configByMethod == null)
+        {
+            synchronized (this)
+            {
+                if (configByMethod == null)
+                {
+                    configByMethod = new ConcurrentHashMap<Method, ExecutorService>();
+                }
+            }
+        }
+
+        // running < j8 we cant have cancellation
+        //final AtomicReference<Callable<?>> cancelHook = new AtomicReference<Callable<?>>();
+        final Callable<Object> invocation = new Callable<Object>()
+        {
+            @Override
+            public Object call() throws Exception
+            {
+                try
+                {
+                    final Object proceed = ic.proceed();
+                    final Future<?> future = !COMPLETION_STAGE.isInstance(proceed) ?
+                            Future.class.cast(proceed) :
+                            Future.class.cast(COMPLETABLE_STAGE_TO_FUTURE.invoke(proceed));
+                    /*
+                    cancelHook.set(new Callable<Boolean>()
+                    {
+                        @Override
+                        public Boolean call()
+                        {
+                            return future.cancel(true);
+                        }
+                    });
+                    */
+                    return future.get();
+                }
+                catch (final InvocationTargetException e)
+                {
+                    throw rethrow(e.getCause());
+                }
+                catch (final Exception e)
+                {
+                    throw rethrow(e);
+                }
+            }
+        };
+
+        final ExecutorService pool = getOrCreatePool(ic);
+        if (COMPLETABLE_FUTURE == null)  // not on java 8 can only be a future
+        {
+            return pool.submit(invocation);
+        }
+
+        // java 8, use CompletableFuture, it impl CompletionStage and Future so everyone is happy
+        final Object completableFuture = COMPLETABLE_FUTURE.newInstance();
+        pool.submit(new J8PromiseCompanionTask(completableFuture, invocation));
+        // TODO: handle cancel
+        return completableFuture;
+    }
+
+    private RuntimeException rethrow(final Throwable cause)
+    {
+        if (RuntimeException.class.isInstance(cause))
+        {
+            return RuntimeException.class.cast(cause);
+        }
+        return new IllegalStateException(cause);
+    }
+
+    private ExecutorService getOrCreatePool(final InvocationContext ic)
+    {
+        final Method method = ic.getMethod();
+        ExecutorService executorService = configByMethod.get(method);
+        if (executorService == null)
+        {
+            final AnnotatedType<?> annotatedType = beanManager.createAnnotatedType(method.getDeclaringClass());
+            AnnotatedMethod<?> annotatedMethod = null;
+            for (final AnnotatedMethod<?> am : annotatedType.getMethods())
+            {
+                if (am.getJavaMember().equals(method))
+                {
+                    annotatedMethod = am;
+                    break;
+                }
+            }
+            if (annotatedMethod == null)
+            {
+                throw new IllegalStateException("No annotated method for " + method);
+            }
+            final Futureable methodConfig = annotatedMethod.getAnnotation(Futureable.class);
+            final ExecutorService instance = manager.find(
+                    (methodConfig == null ? annotatedType.getAnnotation(Futureable.class) : methodConfig).value());
+            configByMethod.putIfAbsent(method, instance);
+            executorService = instance;
+        }
+        return executorService;
+    }
+
+    @ApplicationScoped
+    public static class ThreadPoolManager
+    {
+        private final ConcurrentMap<String, ExecutorService> pools = new ConcurrentHashMap<String, ExecutorService>();
+        private volatile ExecutorService defaultPool;
+        private volatile boolean closed = false;
+
+        @PreDestroy
+        private void shutdown()
+        {
+            closed = true;
+            final String propertyValue = ConfigResolver.getPropertyValue("deltaspike.future.timeout");
+            final long timeout = propertyValue == null ? TimeUnit.MINUTES.toMillis(1) : Integer.parseInt(propertyValue);
+            for (final ExecutorService es : pools.values())
+            {
+                es.shutdown();
+                try
+                {
+                    es.awaitTermination(timeout, TimeUnit.MILLISECONDS);
+                }
+                catch (final InterruptedException e)
+                {
+                    Thread.interrupted();
+                }
+            }
+            if (defaultPool != null)
+            {
+                defaultPool.shutdown();
+                try
+                {
+                    defaultPool.awaitTermination(timeout, TimeUnit.MILLISECONDS);
+                }
+                catch (final InterruptedException e)
+                {
+                    Thread.interrupted();
+                }
+            }
+            pools.clear();
+        }
+
+        // open door for users until we have a config, should be part of API but since it can change keeping it there
+        public void register(final String name, final ExecutorService es)
+        {
+            pools.putIfAbsent(name, es);
+        }
+
+        public ExecutorService find(final String name)
+        {
+            if (closed)
+            {
+                throw new IllegalStateException("Container is shutting down");
+            }
+            ExecutorService pool = pools.get(name);
+            if (pool == null)
+            {
+                ensureDefaultPool();
+                pool = defaultPool;
+            }
+            return pool;
+        }
+
+        private void ensureDefaultPool()
+        {
+            if (defaultPool == null)
+            {
+                synchronized (this)
+                {
+                    if (defaultPool == null)
+                    {
+                        defaultPool = Executors.newFixedThreadPool(
+                                Math.max(2, Runtime.getRuntime().availableProcessors()));
+                    }
+                }
+            }
+        }
+    }
+
+    private static final class J8PromiseCompanionTask<T> implements Runnable
+    {
+        private Object dep;
+        private Callable<T> fn;
+
+        J8PromiseCompanionTask(final Object dep, Callable<T> fn)
+        {
+            this.dep = dep;
+            this.fn = fn;
+        }
+
+        public void run()
+        {
+            try
+            {
+                COMPLETABLE_FUTURE_COMPLETE.invoke(dep, fn.call());
+            }
+            catch (final InvocationTargetException e)
+            {
+                try
+                {
+                    COMPLETABLE_FUTURE_COMPLETE_ERROR.invoke(dep, e.getCause());
+                }
+                catch (IllegalAccessException e1)
+                {
+                    throw new IllegalStateException(e1);
+                }
+                catch (final InvocationTargetException e1)
+                {
+                    throw new IllegalStateException(e1.getCause());
+                }
+            }
+            catch (Exception e)
+            {
+                try
+                {
+                    COMPLETABLE_FUTURE_COMPLETE_ERROR.invoke(dep, e);
+                }
+                catch (IllegalAccessException e1)
+                {
+                    throw new IllegalStateException(e1);
+                }
+                catch (final InvocationTargetException e1)
+                {
+                    throw new IllegalStateException(e1.getCause());
+                }
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/throttling/ThrottledInterceptor.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/throttling/ThrottledInterceptor.java b/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/throttling/ThrottledInterceptor.java
new file mode 100644
index 0000000..6b9e1f8
--- /dev/null
+++ b/deltaspike/core/impl/src/main/java/org/apache/deltaspike/core/impl/throttling/ThrottledInterceptor.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.deltaspike.core.impl.throttling;
+
+import org.apache.deltaspike.core.api.throttling.Throttled;
+import org.apache.deltaspike.core.api.throttling.Throttling;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.enterprise.inject.Typed;
+import javax.enterprise.inject.spi.AnnotatedMethod;
+import javax.enterprise.inject.spi.AnnotatedType;
+import javax.enterprise.inject.spi.BeanManager;
+import javax.inject.Inject;
+import javax.interceptor.AroundInvoke;
+import javax.interceptor.Interceptor;
+import javax.interceptor.InvocationContext;
+import java.io.Serializable;
+import java.lang.reflect.Method;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+@Throttled
+@Interceptor
+public class ThrottledInterceptor implements Serializable
+{
+    @Inject
+    private LocalCache metadata;
+
+    @AroundInvoke
+    public Object invoke(final InvocationContext ic) throws Exception
+    {
+        return metadata.getOrCreateInvocation(ic).invoke(ic);
+    }
+
+    private static Semaphore onInterruption(final InterruptedException e)
+    {
+        Thread.interrupted();
+        throw new IllegalStateException("acquire() interrupted", e);
+    }
+
+    @ApplicationScoped
+    @Typed(LocalCache.class)
+    static class LocalCache implements Throttling.SemaphoreFactory
+    {
+        private final ConcurrentMap<String, Semaphore> semaphores = new ConcurrentHashMap<String, Semaphore>();
+        private final ConcurrentMap<Method, Invocation> providers = new ConcurrentHashMap<Method, Invocation>();
+
+        @Inject
+        private BeanManager beanManager;
+
+        Invocation getOrCreateInvocation(final InvocationContext ic)
+        {
+            final Method method = ic.getMethod();
+            Invocation i = providers.get(method);
+            if (i == null)
+            {
+                final Class declaringClass = method.getDeclaringClass();
+                final AnnotatedType<Object> annotatedType = beanManager.createAnnotatedType(declaringClass);
+                AnnotatedMethod<?> annotatedMethod = null;
+                for (final AnnotatedMethod<?> am : annotatedType.getMethods())
+                {
+                    if (am.getJavaMember().equals(method))
+                    {
+                        annotatedMethod = am;
+                        break;
+                    }
+                }
+                if (annotatedMethod == null)
+                {
+                    throw new IllegalStateException("No annotated method for " + method);
+                }
+
+                Throttled config = annotatedMethod.getAnnotation(Throttled.class);
+                if (config == null)
+                {
+                    config = annotatedType.getAnnotation(Throttled.class);
+                }
+                Throttling sharedConfig = annotatedMethod.getAnnotation(Throttling.class);
+                if (sharedConfig == null)
+                {
+                    sharedConfig = annotatedType.getAnnotation(Throttling.class);
+                }
+
+                final Throttling.SemaphoreFactory factory =
+                        sharedConfig != null && sharedConfig.factory() != Throttling.SemaphoreFactory.class ?
+                        Throttling.SemaphoreFactory.class.cast(
+                                beanManager.getReference(beanManager.resolve(
+                                        beanManager.getBeans(
+                                                sharedConfig.factory())),
+                                        Throttling.SemaphoreFactory.class, null)) : this;
+
+                final Semaphore semaphore = factory.newSemaphore(
+                        annotatedMethod,
+                        sharedConfig != null && !sharedConfig.name().isEmpty() ?
+                                sharedConfig.name() : declaringClass.getName(),
+                        sharedConfig != null && sharedConfig.fair(),
+                        sharedConfig != null ? sharedConfig.permits() : 1);
+                final long timeout = config.timeoutUnit().toMillis(config.timeout());
+                final int weigth = config.weight();
+                i = new Invocation(semaphore, weigth, timeout);
+                final Invocation existing = providers.putIfAbsent(ic.getMethod(), i);
+                if (existing != null)
+                {
+                    i = existing;
+                }
+            }
+            return i;
+        }
+
+        @Override
+        public Semaphore newSemaphore(final AnnotatedMethod<?> method, final String name,
+                                      final boolean fair, final int permits)
+        {
+            Semaphore semaphore = semaphores.get(name);
+            if (semaphore == null)
+            {
+                semaphore = new Semaphore(permits, fair);
+                final Semaphore existing = semaphores.putIfAbsent(name, semaphore);
+                if (existing != null)
+                {
+                    semaphore = existing;
+                }
+            }
+            return semaphore;
+        }
+    }
+
+    private static final class Invocation
+    {
+        private final int weight;
+        private final Semaphore semaphore;
+        private final long timeout;
+
+        private Invocation(final Semaphore semaphore, final int weight, final long timeout)
+        {
+            this.semaphore = semaphore;
+            this.weight = weight;
+            this.timeout = timeout;
+        }
+
+        Object invoke(final InvocationContext context) throws Exception
+        {
+            if (timeout > 0)
+            {
+                try
+                {
+                    if (!semaphore.tryAcquire(weight, timeout, TimeUnit.MILLISECONDS))
+                    {
+                        throw new IllegalStateException("Can't acquire " + weight +
+                                " permits for " + context.getMethod() + " in " + timeout + "ms");
+                    }
+                }
+                catch (final InterruptedException e)
+                {
+                    return onInterruption(e);
+                }
+            }
+            else
+            {
+                try
+                {
+                    semaphore.acquire(weight);
+                }
+                catch (final InterruptedException e)
+                {
+                    return onInterruption(e);
+                }
+            }
+            try
+            {
+                return context.proceed();
+            }
+            finally
+            {
+                semaphore.release(weight);
+            }
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/impl/src/main/resources/META-INF/beans.xml
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/main/resources/META-INF/beans.xml b/deltaspike/core/impl/src/main/resources/META-INF/beans.xml
index a2308ff..9994be5 100644
--- a/deltaspike/core/impl/src/main/resources/META-INF/beans.xml
+++ b/deltaspike/core/impl/src/main/resources/META-INF/beans.xml
@@ -20,4 +20,8 @@
 <beans xmlns="http://java.sun.com/xml/ns/javaee"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
        xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/beans_1_0.xsd">
+  <interceptors>
+    <class>org.apache.deltaspike.core.impl.throttling.ThrottledInterceptor</class>
+    <class>org.apache.deltaspike.core.impl.future.FutureableInterceptor</class>
+  </interceptors>
 </beans>

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/future/FutureableTest.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/future/FutureableTest.java b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/future/FutureableTest.java
new file mode 100644
index 0000000..afc37e7
--- /dev/null
+++ b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/future/FutureableTest.java
@@ -0,0 +1,66 @@
+package org.apache.deltaspike.test.core.impl.future;
+
+import org.apache.deltaspike.test.util.ArchiveUtils;
+import org.jboss.arquillian.container.test.api.Deployment;
+import org.jboss.arquillian.junit.Arquillian;
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.asset.EmptyAsset;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.jboss.shrinkwrap.api.spec.WebArchive;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import javax.inject.Inject;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+@RunWith(Arquillian.class)
+public class FutureableTest {
+    @Deployment
+    public static WebArchive deploy()
+    {
+        JavaArchive testJar = ShrinkWrap.create(JavaArchive.class, "FutureableTest.jar")
+                .addPackage(Service.class.getPackage().getName())
+                .addAsManifestResource(EmptyAsset.INSTANCE, "beans.xml");
+
+        return ShrinkWrap.create(WebArchive.class, "FutureableTest.war")
+                .addAsLibraries(ArchiveUtils.getDeltaSpikeCoreArchive())
+                .addAsLibraries(testJar)
+                .addAsWebInfResource(EmptyAsset.INSTANCE, "beans.xml");
+    }
+
+    @Inject
+    private Service service;
+
+    @Test
+    public void future()
+    {
+        final Future<String> future = service.thatSLong(1000);
+        int count = 0;
+        for (int i = 0; i < 1000; i++)
+        {
+            if (future.isDone())
+            {
+                break;
+            }
+            count++;
+        }
+        try
+        {
+            assertEquals("done", future.get());
+        }
+        catch (final InterruptedException e)
+        {
+            Thread.interrupted();
+            fail();
+        }
+        catch (final ExecutionException e)
+        {
+            fail(e.getMessage());
+        }
+        assertEquals(1000, count);
+    }
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/future/Service.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/future/Service.java b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/future/Service.java
new file mode 100644
index 0000000..2212302
--- /dev/null
+++ b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/future/Service.java
@@ -0,0 +1,61 @@
+package org.apache.deltaspike.test.core.impl.future;
+
+import org.apache.deltaspike.core.api.future.Futureable;
+
+import javax.enterprise.context.ApplicationScoped;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+@ApplicationScoped
+public class Service
+{
+    @Futureable // or CompletableFuture<String>
+    public Future<String> thatSLong(final long sleep)
+    {
+        try
+        {
+            Thread.sleep(sleep);
+            // return CompletableFuture.completedFuture("done");
+            return new Future<String>()  // EE will have AsyncFuture but more designed for j8 ^^
+            {
+                @Override
+                public boolean cancel(final boolean mayInterruptIfRunning)
+                {
+                    return false;
+                }
+
+                @Override
+                public boolean isCancelled()
+                {
+                    return false;
+                }
+
+                @Override
+                public boolean isDone()
+                {
+                    return true;
+                }
+
+                @Override
+                public String get() throws InterruptedException, ExecutionException
+                {
+                    return "done";
+                }
+
+                @Override
+                public String get(final long timeout, final TimeUnit unit)
+                        throws InterruptedException, ExecutionException, TimeoutException
+                {
+                    return "done";
+                }
+            };
+        }
+        catch (final InterruptedException e)
+        {
+            Thread.interrupted();
+            throw new IllegalStateException(e);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/Service.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/Service.java b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/Service.java
new file mode 100644
index 0000000..95f7e7f
--- /dev/null
+++ b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/Service.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.deltaspike.test.core.impl.throttling;
+
+import org.apache.deltaspike.core.api.throttling.Throttled;
+
+import javax.enterprise.context.ApplicationScoped;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.fail;
+
+@ApplicationScoped
+public class Service
+{
+    private final Map<String, String> entries = new HashMap<String, String>();
+
+    @Throttled(timeout = 1, timeoutUnit = TimeUnit.SECONDS)
+    public String read(final String k)
+    {
+        return entries.get(k);
+    }
+
+    @Throttled(timeout = 1, timeoutUnit = TimeUnit.SECONDS)
+    public void write(final String k, final String v)
+    {
+        entries.put(k, v);
+    }
+
+    @Throttled
+    public void force()
+    {
+        try
+        {
+            Thread.sleep(TimeUnit.SECONDS.toMillis(5));
+        }
+        catch (final InterruptedException e)
+        {
+            Thread.interrupted();
+            fail();
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/Service2.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/Service2.java b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/Service2.java
new file mode 100644
index 0000000..4d81d51
--- /dev/null
+++ b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/Service2.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.apache.deltaspike.test.core.impl.throttling;
+
+import org.apache.deltaspike.core.api.throttling.Throttled;
+import org.apache.deltaspike.core.api.throttling.Throttling;
+
+import javax.enterprise.context.ApplicationScoped;
+import java.util.ArrayList;
+import java.util.Collection;
+
+@Throttling(permits = 2)
+@ApplicationScoped
+public class Service2
+{
+    private final Collection<String> called = new ArrayList<String>();
+
+    @Throttled(timeout = 750)
+    public void call(final String k)
+    {
+        synchronized (called)
+        {
+            called.add(k);
+        }
+        try
+        {
+            Thread.sleep(1000);
+        }
+        catch (final InterruptedException e)
+        {
+            Thread.interrupted();
+        }
+    }
+
+    @Throttled(weight = 2)
+    public void heavy(final Runnable inTask)
+    {
+        inTask.run();
+        try
+        {
+            Thread.sleep(5000);
+        }
+        catch (final InterruptedException e)
+        {
+            Thread.interrupted();
+        }
+    }
+
+    public Collection<String> getCalled()
+    {
+        return called;
+    }
+}

http://git-wip-us.apache.org/repos/asf/deltaspike/blob/dbd4fb78/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/ThrottledTest.java
----------------------------------------------------------------------
diff --git a/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/ThrottledTest.java b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/ThrottledTest.java
new file mode 100644
index 0000000..261a434
--- /dev/null
+++ b/deltaspike/core/impl/src/test/java/org/apache/deltaspike/test/core/impl/throttling/ThrottledTest.java
@@ -0,0 +1,277 @@
+package org.apache.deltaspike.test.core.impl.throttling;
+
+import org.apache.deltaspike.test.category.SeCategory;
+import org.apache.deltaspike.test.util.ArchiveUtils;
+import org.jboss.arquillian.container.test.api.Deployment;
+import org.jboss.arquillian.junit.Arquillian;
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.asset.EmptyAsset;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.jboss.shrinkwrap.api.spec.WebArchive;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+
+import javax.inject.Inject;
+import java.util.HashSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static java.util.Arrays.asList;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
+
+@RunWith(Arquillian.class)
+@Category(SeCategory.class)
+public class ThrottledTest {
+    @Deployment
+    public static WebArchive deploy()
+    {
+        JavaArchive testJar = ShrinkWrap.create(JavaArchive.class, "ThrottledTest.jar")
+                .addPackage(Service.class.getPackage().getName())
+                .addAsManifestResource(EmptyAsset.INSTANCE, "beans.xml");
+
+        return ShrinkWrap.create(WebArchive.class, "ThrottledTest.war")
+                .addAsLibraries(ArchiveUtils.getDeltaSpikeCoreArchive())
+                .addAsLibraries(testJar)
+                .addAsWebInfResource(EmptyAsset.INSTANCE, "beans.xml");
+    }
+
+    @Inject
+    private Service service;
+
+    @Inject
+    private Service2 service2;
+
+    @Test
+    public void permits()
+    {
+        {// failling case now
+            final AtomicReference<Exception> failed = new AtomicReference<Exception>();
+            final CountDownLatch latch = new CountDownLatch(2);
+            final Thread[] concurrents = new Thread[]
+                    {
+                    new Thread()
+                    {
+                        @Override
+                        public void run()
+                        {
+                            service2.heavy(new Runnable()
+                            {
+                                @Override
+                                public void run()
+                                {
+                                    latch.countDown();
+                                }
+                            });
+                        }
+                    },
+                    new Thread()
+                    {
+                        @Override
+                        public void run()
+                        {
+                            try
+                            {
+                                latch.await();
+                            }
+                            catch (final InterruptedException e)
+                            {
+                                Thread.interrupted();
+                                fail();
+                            }
+                            try
+                            {
+                                service2.call("failed");
+                                fail();
+                            }
+                            catch (final IllegalStateException ise)
+                            {
+                                failed.set(ise);
+                            }
+                        }
+                    }
+            };
+            for (final Thread t : concurrents)
+            {
+                t.start();
+            }
+            latch.countDown();
+            waitForThreads(concurrents);
+            assertNotNull(failed.get());
+            assertThat(failed.get(), instanceOf(IllegalStateException.class));
+        }
+        { // passing
+            final CountDownLatch latch = new CountDownLatch(1);
+            final Thread[] concurrents = new Thread[]
+                    {
+                    new Thread()
+                    {
+                        @Override
+                        public void run()
+                        {
+                            try
+                            {
+                                latch.await();
+                            }
+                            catch (final InterruptedException e)
+                            {
+                                Thread.interrupted();
+                                fail();
+                            }
+                            service2.call("1");
+                        }
+                    },
+                    new Thread()
+                    {
+                        @Override
+                        public void run()
+                        {
+                            try
+                            {
+                                latch.await();
+                            }
+                            catch (final InterruptedException e)
+                            {
+                                Thread.interrupted();
+                                fail();
+                            }
+                            service2.call("2");
+                        }
+                    }
+            };
+            for (final Thread t : concurrents)
+            {
+                t.start();
+            }
+            latch.countDown();
+            waitForThreads(concurrents);
+            assertEquals(new HashSet<String>(asList("1", "2")), new HashSet<String>(service2.getCalled()));
+        }
+    }
+
+    private void waitForThreads(final Thread[] concurrents)
+    {
+        for (final Thread t : concurrents)
+        {
+            try
+            {
+                t.join();
+            }
+            catch (final InterruptedException e)
+            {
+                Thread.interrupted();
+                fail();
+            }
+        }
+    }
+
+    @Test
+    public void simpleNotConcurrent()
+    { // ~lock case
+        final CountDownLatch synchro = new CountDownLatch(1);
+        final Thread writer = new Thread()
+        {
+            @Override
+            public void run()
+            {
+                service.write("test", "value");
+                synchro.countDown();
+            }
+        };
+
+        final CountDownLatch end = new CountDownLatch(1);
+        final AtomicReference<String> val = new AtomicReference<String>();
+        final Thread reader = new Thread()
+        {
+            @Override
+            public void run()
+            {
+                try
+                {
+                    synchro.await(1, TimeUnit.MINUTES);
+                }
+                catch (final InterruptedException e)
+                {
+                    Thread.interrupted();
+                    fail();
+                }
+                val.set(service.read("test"));
+                end.countDown();
+            }
+        };
+
+        reader.start();
+        writer.start();
+        try
+        {
+            end.await(1, TimeUnit.MINUTES);
+        }
+        catch (final InterruptedException e)
+        {
+            Thread.interrupted();
+            fail();
+        }
+        assertEquals("value", val.get());
+    }
+
+    @Test
+    public void concurrentTimeout()
+    {
+        final AtomicBoolean doAgain = new AtomicBoolean(true);
+        final CountDownLatch endWriter = new CountDownLatch(1);
+        final Thread writer = new Thread()
+        {
+            @Override
+            public void run()
+            {
+                while (doAgain.get())
+                {
+                    service.write("test", "value");
+                    service.force();
+                }
+                endWriter.countDown();
+            }
+        };
+
+        final CountDownLatch endReader = new CountDownLatch(1);
+        final Thread reader = new Thread()
+        {
+            @Override
+            public void run()
+            {
+                while (doAgain.get())
+                {
+                    try
+                    {
+                        service.read("test");
+                    }
+                    catch (final IllegalStateException e)
+                    {
+                        doAgain.set(false);
+                    }
+                }
+                endReader.countDown();
+            }
+        };
+
+        reader.start();
+        writer.start();
+        try
+        {
+            endReader.await(1, TimeUnit.MINUTES);
+            endWriter.await(1, TimeUnit.MINUTES);
+        }
+        catch (final InterruptedException e)
+        {
+            Thread.interrupted();
+            fail();
+        }
+        assertEquals("value", service.read("test"));
+    }
+}


Mime
View raw message