From 132aa8a77685ec92bc90c03f987650d275a7b639 Mon Sep 17 00:00:00 2001 From: lresende Date: Mon, 30 Sep 2013 06:59:11 +0000 Subject: 2.0.1 RC1 release tag git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@1527464 13f79535-47bb-0310-9956-ffa450edef68 --- .../sca/core/work/impl/DefaultWorkScheduler.java | 210 +++++++++++++++++++++ 1 file changed, 210 insertions(+) create mode 100644 sca-java-2.x/tags/2.0.1-RC1/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/DefaultWorkScheduler.java (limited to 'sca-java-2.x/tags/2.0.1-RC1/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/DefaultWorkScheduler.java') diff --git a/sca-java-2.x/tags/2.0.1-RC1/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/DefaultWorkScheduler.java b/sca-java-2.x/tags/2.0.1-RC1/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/DefaultWorkScheduler.java new file mode 100644 index 0000000000..dd4abf0cd2 --- /dev/null +++ b/sca-java-2.x/tags/2.0.1-RC1/modules/core/src/main/java/org/apache/tuscany/sca/core/work/impl/DefaultWorkScheduler.java @@ -0,0 +1,210 @@ +/* + * 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.tuscany.sca.core.work.impl; + +import java.security.AccessController; +import java.security.PrivilegedAction; +import java.util.Map; +import java.util.concurrent.ExecutorService; + +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +import org.apache.tuscany.sca.core.LifeCycleListener; +import org.apache.tuscany.sca.work.NotificationListener; +import org.apache.tuscany.sca.work.WorkScheduler; +import org.apache.tuscany.sca.work.WorkSchedulerException; + +/** + * A work scheduler implementation based on a JSR 237 work manager. + *

+ *

+ * This needs a JSR 237 work manager implementation available for scheduling work. Instances can be configured with a + * work manager implementation that is injected in. It is the responsibility of the runtime environment to make a work + * manager implementation available. For example, if the managed environment supports work manager the runtime can use + * the appropriate lookup mechanism to inject the work manager implementation.

+ * + * @version $Rev$ $Date$ + */ +public class DefaultWorkScheduler implements WorkScheduler, LifeCycleListener { + + /** + * Underlying JSR-237 work manager + */ + private ThreadPoolWorkManager jsr237WorkManager; + private int maxThreads = 0; + + /** + * Initializes the JSR 237 work manager. + * + * @param jsr237WorkManager JSR 237 work manager. + */ + public DefaultWorkScheduler(ExtensionPointRegistry registry, Map attributes) { + if (attributes != null) { + String value = attributes.get("maxThreads"); + if (value != null) { + maxThreads = Integer.parseInt(value.trim()); + } + } + } + + private synchronized ThreadPoolWorkManager getWorkManager() { + if (jsr237WorkManager != null) { + return jsr237WorkManager; + } +// try { +// InitialContext ctx = new InitialContext(); +// jsr237WorkManager = (ThreadPoolWorkManager)ctx.lookup("java:comp/env/wm/TuscanyWorkManager"); +// } catch (Throwable e) { +// // ignore +// } + if (jsr237WorkManager == null) { + jsr237WorkManager = new ThreadPoolWorkManager(maxThreads); + } + return jsr237WorkManager; + } + + /** + * Schedules a unit of work for future execution. The notification listener is used to register interest in + * callbacks regarding the status of the work. + * + * @param work The unit of work that needs to be asynchronously executed. + */ + public void scheduleWork(T work) { + scheduleWork(work, null); + } + + /** + * Schedules a unit of work for future execution. The notification listener is used to register interest in + * callbacks regarding the status of the work. + * + * @param work The unit of work that needs to be asynchronously executed. + * @param listener Notification listener for callbacks. + */ + public void scheduleWork(T work, NotificationListener listener) { + + if (work == null) { + throw new IllegalArgumentException("Work cannot be null"); + } + + Work jsr237Work = new Work(work); + try { + if (listener == null) { + getWorkManager().schedule(jsr237Work); + } else { + Jsr237WorkListener jsr237WorkListener = new Jsr237WorkListener(listener, work); + getWorkManager().schedule(jsr237Work, jsr237WorkListener); + } + } catch (IllegalArgumentException ex) { + if (listener != null) { + listener.workRejected(work); + } else { + throw new WorkSchedulerException(ex); + } + } catch (Exception ex) { + throw new WorkSchedulerException(ex); + } + + } + + public void start() { + } + + public void stop() { + if (jsr237WorkManager instanceof ThreadPoolWorkManager) { + // Allow privileged access to modify threads. Requires RuntimePermission in security + // policy. + AccessController.doPrivileged(new PrivilegedAction() { + public Object run() { + ((ThreadPoolWorkManager)jsr237WorkManager).destroy(); + return null; + } + }); + } + } + + /* + * WorkListener for keeping track of work status callbacks. + * + */ + private class Jsr237WorkListener implements WorkListener { + + // Notification listener + private NotificationListener listener; + + // Work + private T work; + + /* + * Initializes the notification listener. + */ + public Jsr237WorkListener(NotificationListener listener, T work) { + this.listener = listener; + this.work = work; + } + + /* + * Callback when the work is accepted. + */ + public void workAccepted(WorkEvent workEvent) { + T work = getWork(); + listener.workAccepted(work); + } + + /* + * Callback when the work is rejected. + */ + public void workRejected(WorkEvent workEvent) { + T work = getWork(); + listener.workRejected(work); + } + + /* + * Callback when the work is started. + */ + public void workStarted(WorkEvent workEvent) { + T work = getWork(); + listener.workStarted(work); + } + + /* + * Callback when the work is completed. + */ + public void workCompleted(WorkEvent workEvent) { + T work = getWork(); + Exception exception = workEvent.getException(); + if (exception != null) { + listener.workFailed(work, exception); + } else { + listener.workCompleted(work); + } + } + + /* + * Gets the underlying work from the work event. + */ + private T getWork() { + return work; + } + + } + + @Override + public ExecutorService getExecutorService() { + return getWorkManager().getExecutorService(); + } +} -- cgit v1.2.3