diff options
Diffstat (limited to 'sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work')
5 files changed, 713 insertions, 0 deletions
diff --git a/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/Jsr237Work.java b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/Jsr237Work.java new file mode 100644 index 0000000000..6fe883157f --- /dev/null +++ b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/Jsr237Work.java @@ -0,0 +1,63 @@ +/* + * 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; + +/* + * JCA work wrapper. + */ +public class Jsr237Work<T extends Runnable> implements commonj.work.Work { + + // Work that is being executed. + private T work; + + /* + * Initializes the work instance. + */ + public Jsr237Work(T work) { + this.work = work; + } + + /* + * Returns the completed work. + */ + public T getWork() { + return work; + } + + /* + * Release the work. + */ + public void release() { + } + + /* + * Work attributes are not daemon. + */ + public boolean isDaemon() { + return false; + } + + /* + * Runs the work. + */ + public void run() { + work.run(); + } +}
\ No newline at end of file diff --git a/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/Jsr237WorkScheduler.java b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/Jsr237WorkScheduler.java new file mode 100644 index 0000000000..ea213f5311 --- /dev/null +++ b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/Jsr237WorkScheduler.java @@ -0,0 +1,180 @@ +/* + * 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; + +import javax.naming.InitialContext; +import javax.naming.NamingException; + +import org.apache.tuscany.sca.work.NotificationListener; +import org.apache.tuscany.sca.work.WorkScheduler; +import org.apache.tuscany.sca.work.WorkSchedulerException; + +import commonj.work.WorkEvent; +import commonj.work.WorkListener; +import commonj.work.WorkManager; + +/** + * A work scheduler implementation based on a JSR 237 work manager. + * <p/> + * <p/> + * 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 implementaion available. For example, if the managed environment supports work manager the runtime can use + * the appropriate lookup mechanism to inject the work manager implementation. </p> + */ +public class Jsr237WorkScheduler implements WorkScheduler { + + /** + * Underlying JSR-237 work manager + */ + private WorkManager jsr237WorkManager; + + /** + * Initializes the JSR 237 work manager. + * + * @param jsr237WorkManager JSR 237 work manager. + */ + public Jsr237WorkScheduler() { + + try { + InitialContext ctx = new InitialContext(); + jsr237WorkManager = (WorkManager) ctx.lookup("java:comp/env/wm/TuscanyWorkManager"); + } catch (NamingException e) { + // ignore + } + if (jsr237WorkManager == null) { + jsr237WorkManager = new ThreadPoolWorkManager(10); + } + } + + /** + * 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 <T extends Runnable> 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 <T extends Runnable> void scheduleWork(T work, NotificationListener<T> listener) { + + if (work == null) { + throw new IllegalArgumentException("Work cannot be null"); + } + + Jsr237Work<T> jsr237Work = new Jsr237Work<T>(work); + try { + if (listener == null) { + jsr237WorkManager.schedule(jsr237Work); + } else { + Jsr237WorkListener<T> jsr237WorkListener = new Jsr237WorkListener<T>(listener, work); + jsr237WorkManager.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 destroy() { + if (jsr237WorkManager instanceof ThreadPoolWorkManager) { + ((ThreadPoolWorkManager)jsr237WorkManager).destroy(); + } + } + + /* + * Worklistener for keeping track of work status callbacks. + * + */ + private class Jsr237WorkListener<T extends Runnable> implements WorkListener { + + // Notification listener + private NotificationListener<T> listener; + + // Work + private T work; + + /* + * Initializes the notification listener. + */ + public Jsr237WorkListener(NotificationListener<T> 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; + } + + } +} diff --git a/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/ThreadPoolWorkManager.java b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/ThreadPoolWorkManager.java new file mode 100644 index 0000000000..7da779bc21 --- /dev/null +++ b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/ThreadPoolWorkManager.java @@ -0,0 +1,228 @@ +/* + * 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; + +import java.rmi.server.UID; +import java.util.Collection; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ThreadFactory; + +import org.osoa.sca.annotations.Destroy; + +import commonj.work.Work; +import commonj.work.WorkEvent; +import commonj.work.WorkException; +import commonj.work.WorkItem; +import commonj.work.WorkListener; +import commonj.work.WorkManager; + +/** + * A thread-pool based implementation for the JSR-237 work manager. + * <p/> + * <p/> + * This implementation supports only local work. + * <p/> + * TODO Elaborate the implementation. </p> + */ +public class ThreadPoolWorkManager implements WorkManager { + + // Map of work items currently handled by the work manager + private Map<WorkItemImpl, WorkListener> workItems = new ConcurrentHashMap<WorkItemImpl, WorkListener>(); + + // Thread-pool + private ExecutorService executor; + + /** + * Initializes the thread-pool. + * + * @param threadPoolSize Thread-pool size. + */ + public ThreadPoolWorkManager(int threadPoolSize) { + + // Creates a new Executor, use a custom ThreadFactory that + // creates daemon threads. + executor = Executors.newFixedThreadPool(threadPoolSize, new ThreadFactory() { + public Thread newThread(Runnable r) { + Thread thread = new Thread(r); + thread.setDaemon(true); + return thread; + } + }); + } + + /** + * Schedules a unit of work asynchronously. + * + * @param work Work that needs to be scheduled. + * @return Work Work item representing the asynchronous work + */ + public WorkItem schedule(Work work) throws IllegalArgumentException { + return schedule(work, null); + } + + /** + * Schedules a unit of work asynchronously. + * + * @param work Work that needs to be scheduled. + * @param workListener Work listener for callbacks. + * @return Work Work item representing the asynchronous work + */ + public WorkItem schedule(Work work, WorkListener workListener) throws IllegalArgumentException { + + WorkItemImpl workItem = new WorkItemImpl(new UID().toString(), work); + if (workListener != null) { + workItems.put(workItem, workListener); + } + workAccepted(workItem, work); + if (scheduleWork(work, workItem)) { + return workItem; + } else { + workItem.setStatus(WorkEvent.WORK_REJECTED); + if (workListener != null) { + workListener.workRejected(new WorkEventImpl(workItem)); + } + throw new IllegalArgumentException("Unable to schedule work"); + } + } + + /** + * Wait for all the specified units of work to finish. + * + * @param works Units of the work that need to finish. + * @param timeout Timeout for waiting for the units of work to finish. + */ + public boolean waitForAll(Collection works, long timeout) { + throw new UnsupportedOperationException("waitForAll not supported"); + } + + /** + * Wait for any of the specified units of work to finish. + * + * @param works Units of the work that need to finish. + * @param timeout Timeout for waiting for the units of work to finish. + */ + public Collection waitForAny(Collection works, long timeout) { + throw new UnsupportedOperationException("waitForAny not supported"); + } + + /** + * Method provided for subclasses to indicate a work accptance. + * + * @param workItem Work item representing the work that was accepted. + * @param work Work that was accepted. + */ + private void workAccepted(final WorkItemImpl workItem, final Work work) { + WorkListener listener = workItems.get(workItem); + if (listener != null) { + workItem.setStatus(WorkEvent.WORK_ACCEPTED); + WorkEvent event = new WorkEventImpl(workItem); + listener.workAccepted(event); + } + } + + /* + * Method to indicate a work start. + */ + private void workStarted(final WorkItemImpl workItem, final Work work) { + WorkListener listener = workItems.get(workItem); + if (listener != null) { + workItem.setStatus(WorkEvent.WORK_STARTED); + WorkEvent event = new WorkEventImpl(workItem); + listener.workStarted(event); + } + } + + /* + * Method to indicate a work completion. + */ + private void workCompleted(final WorkItemImpl workItem, final Work work) { + workCompleted(workItem, work, null); + } + + /* + * Method to indicate a work completion. + */ + private void workCompleted(final WorkItemImpl workItem, final Work work, final WorkException exception) { + WorkListener listener = workItems.get(workItem); + if (listener != null) { + workItem.setStatus(WorkEvent.WORK_COMPLETED); + workItem.setResult(work); + workItem.setException(exception); + WorkEvent event = new WorkEventImpl(workItem); + listener.workCompleted(event); + workItems.remove(workItem); + } + } + + /* + * Schedules the work using the threadpool. + */ + private boolean scheduleWork(final Work work, final WorkItemImpl workItem) { + try { + executor.execute(new DecoratingWork(workItem, work)); + return true; + } catch (RejectedExecutionException ex) { + return false; + } + } + + /* + * Class that decorates the original worker so that it can get callbacks when work is done. + */ + private final class DecoratingWork implements Runnable { + + // Work item for this work. + private WorkItemImpl workItem; + + // The original work. + private Work decoratedWork; + + /* + * Initializes the work item and underlying work. + */ + private DecoratingWork(final WorkItemImpl workItem, final Work decoratedWork) { + this.workItem = workItem; + this.decoratedWork = decoratedWork; + } + + /* + * Overrides the run method. + */ + public void run() { + workStarted(workItem, decoratedWork); + try { + decoratedWork.run(); + workCompleted(workItem, decoratedWork); + } catch (Throwable th) { + workCompleted(workItem, decoratedWork, new WorkException(th.getMessage(), th)); + } + } + + } + + @Destroy + public void destroy() { + executor.shutdown(); + } + +} diff --git a/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/WorkEventImpl.java b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/WorkEventImpl.java new file mode 100644 index 0000000000..62ebeea1fb --- /dev/null +++ b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/WorkEventImpl.java @@ -0,0 +1,74 @@ +/* + * 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; + + +import commonj.work.WorkEvent; +import commonj.work.WorkException; +import commonj.work.WorkItem; + +/** + * Default immutable implementation of the <code>WorkEvent</code> class. + */ +class WorkEventImpl implements WorkEvent { + + // Work item for this event + private WorkItem workItem; + + // Exception if something has gone wrong + private WorkException exception; + + /** + * Instantiates the event. + * + * @param workItem Work item for this event. + */ + public WorkEventImpl(final WorkItemImpl workItem) { + this.workItem = workItem; + this.exception = workItem.getException(); + } + + /** + * Returns the work type based on whether the work was accepted, started, + * rejected or completed. + * + * @return Work type. + */ + public int getType() { + return workItem.getStatus(); + } + + /** + * Returns the work item associated with this work type. + * + * @return Work item. + */ + public WorkItem getWorkItem() { + return workItem; + } + + /** + * Returns the exception if the work completed with an exception. + * + * @return Work exception. + */ + public WorkException getException() { + return exception; + } +} diff --git a/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/WorkItemImpl.java b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/WorkItemImpl.java new file mode 100644 index 0000000000..39a7c647f7 --- /dev/null +++ b/sca-java-1.x/tags/1.2/modules/core/src/main/java/org/apache/tuscany/sca/core/work/WorkItemImpl.java @@ -0,0 +1,168 @@ +/* + * 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; + +import commonj.work.Work; +import commonj.work.WorkException; +import commonj.work.WorkItem; + +/** + * An identity based immutable implementation of the <code>WorkItem</code> + * interface. + * + */ +class WorkItemImpl implements WorkItem { + + // Id scoped for the VM + private String id; + + // Status + private int status = -1; + + // Result + private Work result; + + // Original work + private Work originalWork; + + // Exception + private WorkException exception; + + /** + * Instantiates an id for this item. + * + * @param id of this work event. + */ + protected WorkItemImpl(final String id, final Work orginalWork) { + this.id = id; + this.originalWork = orginalWork; + } + + /** + * Returns the id. + * + * @return Id of this item. + */ + public String getId() { + return id; + } + + /** + * Returns the original work. + * + * @return Original work. + */ + public Work getOriginalWork() { + return originalWork; + } + + /** + * Returns the work result if the work completed. + * + * @return Work. + * @throws WorkException If the work completed with an exception. + */ + public Work getResult() { + return result; + } + + /** + * Sets the result. + * + * @param result Result. + */ + protected void setResult(final Work result) { + this.result = result; + } + + /** + * Returns the exception if work completed with an exception. + * + * @return Work exception. + */ + protected WorkException getException() { + return exception; + } + + /** + * Sets the exception. + * + * @param exception Exception. + */ + protected void setException(final WorkException exception) { + this.exception = exception; + } + + /** + * Returns the work type based on whether the work was accepted, started, + * rejected or completed. + * + * @return Work status. + */ + public int getStatus() { + return status; + } + + /** + * Sets the status. + * + * @param status Status. + */ + protected void setStatus(final int status) { + this.status = status; + } + + /** + * @see Object#hashCode() + */ + @Override + public int hashCode() { + return id.hashCode(); + } + + /** + * Indicates whether some other object is "equal to" this one. + * + * @param obj Object to be compared. + * @return true if this object is the same as the obj argument; false + * otherwise.. + */ + @Override + public boolean equals(final Object obj) { + return (obj != null) && (obj.getClass() == WorkItemImpl.class) && ((WorkItemImpl) obj).id.equals(id); + } + + /** + * Compares this object with the specified object for order. Returns a + * negative integer, zero, or a positive integer as this object is less + * than, equal to, or greater than the specified object. + * + * @param o Object to be compared. + * @return A negative integer, zero, or a positive integer as this object + * is less than, equal to, or greater than the specified object. + * @throws ClassCastException needs better documentation. + */ + public int compareTo(final Object o) { + if (o.getClass() != WorkItemImpl.class) { + throw new ClassCastException(o.getClass().getName()); + } else { + return ((WorkItemImpl) o).getId().compareTo(getId()); + } + } +} |