org.springframework.core.task.TaskRejectedException#javax.resource.spi.work.WorkException源码实例Demo

/**
 * {@inheritDoc}
 */
@Override
public void deltaStartWorkAccepted(Address address)
{
   if (trace)
      log.tracef("DELTA_STARTWORK_ACCEPTED(%s)", address);

   if (address.getTransportId() != null && !getId().equals(address.getTransportId()))
   {
      try
      {
         T addr = nodes.get(address);
         sendMessage(addr, Request.DELTA_STARTWORK_ACCEPTED, address);
      }
      catch (WorkException e1)
      {
         if (log.isDebugEnabled())
         {
            log.debug("Error", e1);
         }
      }
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
public void deltaDoWorkAccepted(Address address)
{
   if (trace)
      log.tracef("DELTA_DOWORK_ACCEPTED(%s)", address);

   if (address.getTransportId() != null && !getId().equals(address.getTransportId()))
   {
      try
      {
         T addr = nodes.get(address);
         sendMessage(addr, Request.DELTA_DOWORK_ACCEPTED, address);
      }
      catch (WorkException e1)
      {
         if (log.isDebugEnabled())
         {
            log.debug("Error", e1);
         }
      }
   }
}
 

@Override
public long getLongRunningFree(Address address)
{
   if (trace)
      log.tracef("GET_LONGRUNNING_FREE(%s)", address);

   if (address.getTransportId() == null || getId().equals(address.getTransportId()))
      return localGetLongRunningFree(address);

   try
   {
      T addr = nodes.get(address);
      return (long)sendMessage(addr, Request.GET_LONGRUNNING_FREE, address);
   }
   catch (WorkException e1)
   {
      if (log.isDebugEnabled())
      {
         log.debug("Error", e1);
      }
      return 0L;
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
public void deltaWorkFailed(Address address)
{
   if (trace)
      log.tracef("DELTA_WORK_FAILED(%s)", address);

   if (address.getTransportId() != null && !getId().equals(address.getTransportId()))
   {
      try
      {
         T addr = nodes.get(address);
         sendMessage(addr, Request.DELTA_WORK_FAILED, address);
      }
      catch (WorkException e1)
      {
         if (log.isDebugEnabled())
         {
            log.debug("Error", e1);
         }
      }
   }
}
 

/**
 * Test for paragraph 5
 * Both the run and release methods in the Work implementation may contain synchronization 
 *            synchronization but they must not be declared as synchronized methods.
 * @throws Throwable throwable exception 
 */
@Test(expected = WorkException.class)
public void testCannotDeclaredSynchronizedRunMethodWork() throws Throwable
{
   SynchronizedRunWork sw = new SynchronizedRunWork();
   WorkConnection wc = wcf.getConnection();
   try
   {
      wc.doWork(sw);
      fail("Synchronized method not catched");
   }
   finally
   {
      wc.close();
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
protected void deltaStartWorkAccepted()
{
   log.trace("deltaStartWorkAccepted");

   super.deltaStartWorkAccepted();

   if (distributedStatisticsEnabled && distributedStatistics != null && transport != null)
   {
      try
      {
         checkTransport();
         distributedStatistics.sendDeltaStartWorkAccepted();
      }
      catch (WorkException we)
      {
         log.debugf("deltaStartWorkAccepted: %s", we.getMessage(), we);
      }
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
protected void deltaScheduleWorkAccepted()
{
   log.trace("deltaScheduleWorkAccepted");

   super.deltaScheduleWorkAccepted();

   if (distributedStatisticsEnabled && distributedStatistics != null && transport != null)
   {
      try
      {
         checkTransport();
         distributedStatistics.sendDeltaScheduleWorkAccepted();
      }
      catch (WorkException we)
      {
         log.debugf("deltaScheduleWorkAccepted: %s", we.getMessage(), we);
      }
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
protected void deltaDoWorkAccepted()
{
   if (trace)
      log.trace("deltaDoWorkAccepted");

   super.deltaDoWorkAccepted();

   if (distributedStatisticsEnabled && distributedStatistics != null && transport != null)
   {
      try
      {
         checkTransport();
         distributedStatistics.sendDeltaDoWorkAccepted();
      }
      catch (WorkException we)
      {
         log.debugf("deltaDoWorkAccepted: %s", we.getMessage(), we);
      }
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
protected void deltaWorkSuccessful()
{
   log.trace("deltaWorkSuccessful");

   super.deltaWorkSuccessful();

   if (distributedStatisticsEnabled && distributedStatistics != null && transport != null)
   {
      try
      {
         checkTransport();
         distributedStatistics.sendDeltaWorkSuccessful();
      }
      catch (WorkException we)
      {
         log.debugf("deltaWorkSuccessful: %s", we.getMessage(), we);
      }
   }
}
 

/**
 * Verify the given work instance.
 * @param work The work
 * @throws WorkException Thrown if a spec compliant issue is found
 */
private void verifyWork(Work work) throws WorkException
{
   Class<? extends Work> workClass = work.getClass();
   String className = workClass.getName();

   if (!validatedWork.contains(className))
   {

      if (isWorkMethodSynchronized(workClass, RUN_METHOD_NAME))
         throw new WorkException(bundle.runMethodIsSynchronized(className));

      if (isWorkMethodSynchronized(workClass, RELEASE_METHOD_NAME))
         throw new WorkException(bundle.releaseMethodIsSynchronized(className));

      validatedWork.add(className);
   }
}
 

/**
 * Check and verify work before submitting.
 * @param work the work instance
 * @param executionContext any execution context that is passed by apadater
 * @throws WorkException if any exception occurs
 */
private void checkAndVerifyWork(Work work, ExecutionContext executionContext) throws WorkException
{
   if (specCompliant)
   {
      verifyWork(work);
   }

   if (work instanceof WorkContextProvider)
   {
      //Implements WorkContextProvider and not-null ExecutionContext
      if (executionContext != null)
      {
         throw new WorkRejectedException(bundle.workExecutionContextMustNullImplementsWorkContextProvider());
      }
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
public long ping(Address address)
{
   log.tracef("PING(%s)", address);

   if (address.getTransportId() == null || getId().equals(address.getTransportId()))
      return localPing();

   long start = System.currentTimeMillis();
   try
   {
      T addr = nodes.get(address);
      sendMessage(addr, Request.PING);
   }
   catch (WorkException e1)
   {
      if (log.isDebugEnabled())
      {
         log.debug("Error", e1);
      }
      return Long.MAX_VALUE;
   }

   return System.currentTimeMillis() - start;
}
 

@Override
public long getLongRunningFree(Address address)
{
   log.tracef("GET_LONGRUNNING_FREE(%s)", address);

   if (address.getTransportId() == null || getId().equals(address.getTransportId()))
      return localGetLongRunningFree(address);

   try
   {
      T addr = nodes.get(address);
      return (long)sendMessage(addr, Request.GET_LONGRUNNING_FREE, address);
   }
   catch (WorkException e1)
   {
      if (log.isDebugEnabled())
      {
         log.debug("Error", e1);
      }
      return 0L;
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
public void deltaDoWorkAccepted(Address address)
{
   log.tracef("DELTA_DOWORK_ACCEPTED(%s)", address);

   if (address.getTransportId() != null && !getId().equals(address.getTransportId()))
   {
      try
      {
         T addr = nodes.get(address);
         sendMessage(addr, Request.DELTA_DOWORK_ACCEPTED, address);
      }
      catch (WorkException e1)
      {
         if (log.isDebugEnabled())
         {
            log.debug("Error", e1);
         }
      }
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
public void deltaScheduleWorkRejected(Address address)
{
   if (trace)
      log.tracef("DELTA_SCHEDULEWORK_REJECTED(%s)", address);

   if (address.getTransportId() != null && !getId().equals(address.getTransportId()))
   {
      try
      {
         T addr = nodes.get(address);
         sendMessage(addr, Request.DELTA_SCHEDULEWORK_REJECTED, address);
      }
      catch (WorkException e1)
      {
         if (log.isDebugEnabled())
         {
            log.debug("Error", e1);
         }
      }
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
protected void deltaStartWorkAccepted()
{
   if (trace)
      log.trace("deltaStartWorkAccepted");

   super.deltaStartWorkAccepted();

   if (distributedStatisticsEnabled && distributedStatistics != null && transport != null)
   {
      try
      {
         checkTransport();
         distributedStatistics.sendDeltaStartWorkAccepted();
      }
      catch (WorkException we)
      {
         log.debugf("deltaStartWorkAccepted: %s", we.getMessage(), we);
      }
   }
}
 

@Override
public long getShortRunningFree(Address address)
{
   if (trace)
      log.tracef("GET_SHORT_RUNNING_FREE(%s)", address);

   if (address.getTransportId() == null || getId().equals(address.getTransportId()))
      return localGetShortRunningFree(address);

   try
   {
      T addr = nodes.get(address);
      return (long)sendMessage(addr, Request.GET_SHORTRUNNING_FREE, address);
   }
   catch (WorkException e1)
   {
      if (log.isDebugEnabled())
      {
         log.debug("Error", e1);
      }
      return 0L;
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
protected void deltaWorkSuccessful()
{
   if (trace)
      log.trace("deltaWorkSuccessful");

   super.deltaWorkSuccessful();

   if (distributedStatisticsEnabled && distributedStatistics != null && transport != null)
   {
      try
      {
         checkTransport();
         distributedStatistics.sendDeltaWorkSuccessful();
      }
      catch (WorkException we)
      {
         log.debugf("deltaWorkSuccessful: %s", we.getMessage(), we);
      }
   }
}
 

/**
 * Check and verify work before submitting.
 * @param work the work instance
 * @param executionContext any execution context that is passed by apadater
 * @throws WorkException if any exception occurs
 */
private void checkAndVerifyWork(Work work, ExecutionContext executionContext) throws WorkException
{
   if (specCompliant)
   {
      verifyWork(work);
   }

   if (work instanceof WorkContextProvider && executionContext != null)
   {
      //Implements WorkContextProvider and not-null ExecutionContext
      throw new WorkRejectedException(bundle.workExecutionContextMustNullImplementsWorkContextProvider());
   }
}
 

/**
 * {@inheritDoc}
 */
public long localStartWork(Work work) throws WorkException
{
   if (transport != null)
   {
      checkTransport();

      if (getLongRunningThreadPool() != null && WorkManagerUtil.isLongRunning(work))
      {
         transport.updateLongRunningFree(getLocalAddress(),
                                         getLongRunningThreadPool().getNumberOfFreeThreads() - 1);
      }
      else
      {
         transport.updateShortRunningFree(getLocalAddress(),
                                          getShortRunningThreadPool().getNumberOfFreeThreads() - 1);
      }

      WorkEventListener wel = new WorkEventListener(WorkManagerUtil.isLongRunning(work),
                                                    getShortRunningThreadPool(),
                                                    getLongRunningThreadPool(),
                                                    getLocalAddress(),
                                                    transport);

      return super.startWork(work, WorkManager.INDEFINITE, null, wel);
   }
   else
   {
      return super.startWork(work);
   }
}
 

@Override
public void scheduleWork(Work work, long startTimeout, @Nullable ExecutionContext executionContext, @Nullable WorkListener workListener)
		throws WorkException {

	Assert.state(this.asyncTaskExecutor != null, "No 'asyncTaskExecutor' set");
	executeWork(this.asyncTaskExecutor, work, startTimeout, false, executionContext, workListener);
}
 

/**
 * localStartWork
 *
 * @param address the logical address
 * @param work the work
 * @return the start value
 * @throws WorkException in case of error
 */
public long localStartWork(Address address, DistributableWork work) throws WorkException
{
   if (trace)
      log.tracef("LOCAL_START_WORK(%s, %s)", address, work);

   DistributedWorkManager dwm = workManagerCoordinator.resolveDistributedWorkManager(address);

   return dwm.localStartWork(work);
}
 

@Override
public void scheduleWork(Work work, long startTimeout, ExecutionContext executionContext, WorkListener workListener)
		throws WorkException {

	Assert.state(this.asyncTaskExecutor != null, "No 'asyncTaskExecutor' set");
	executeWork(this.asyncTaskExecutor, work, startTimeout, false, executionContext, workListener);
}
 

@Override
public void doWork(Work work, long startTimeout, @Nullable ExecutionContext executionContext, @Nullable WorkListener workListener)
		throws WorkException {

	Assert.state(this.syncTaskExecutor != null, "No 'syncTaskExecutor' set");
	executeWork(this.syncTaskExecutor, work, startTimeout, false, executionContext, workListener);
}
 

/**
 * {@inheritDoc}
 */
@Override
public void clearDistributedStatistics(Address address)
{
   if (trace)
      log.tracef("CLEAR_DISTRIBUTED_STATISTICS(%s)", address);

   if (!getId().equals(address.getTransportId()))
      localClearDistributedStatistics(address);

   if (address.getTransportId() != null && getId().equals(address.getTransportId()))
   {
      for (Entry<Address, T> entry : nodes.entrySet())
      {
         Address a = entry.getKey();
         if (!getId().equals(a.getTransportId()))
         {
            try
            {
               sendMessage(entry.getValue(), Request.CLEAR_DISTRIBUTED_STATISTICS, address);
            }
            catch (WorkException e1)
            {
               if (log.isDebugEnabled())
               {
                  log.debug("Error", e1);
               }
            }
         }
      }
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
public void scheduleWork(Address address, DistributableWork work) throws WorkException
{
   if (trace)
      log.tracef("SCHEDULE_WORK(%s, %s)", address, work);

   ClassBundle cb = ClassBundleFactory.createClassBundle(work);

   T addr = nodes.get(address);
   sendMessage(addr, Request.SCHEDULE_WORK, address, cb, work);
}
 

/**
 * {@inheritDoc}
 */
@Override
public void doWork(Work work) throws WorkException
{
   if (policy == null || selector == null || transport == null ||
       work == null || !(work instanceof DistributableWork) || !doWorkDistributionEnabled)
   {
      localDoWork(work);
   }
   else
   {
      doFirstChecks(work, WorkManager.INDEFINITE, null);
      checkTransport();

      DistributableWork dw = (DistributableWork)work;
      boolean executed = false;

      if (policy.shouldDistribute(this, dw))
      {
         Address dwmAddress = selector.selectDistributedWorkManager(getLocalAddress(), dw);
         if (dwmAddress != null && !getLocalAddress().equals(dwmAddress))
         {
            transport.doWork(dwmAddress, dw);
            executed = true;
         }
      }

      if (!executed)
      {
         localDoWork(work);
      }
   }
}
 

/**
 * Execute doWork
 * @param logicalAddress The logical address
 * @param classBundle The class bundle
 * @param b The bytes
 * @throws WorkException in case of error
 */
public void executeDoWork(org.ironjacamar.core.spi.workmanager.Address logicalAddress,
                          ClassBundle classBundle, byte[] b)
   throws WorkException
{
   ByteArrayInputStream bias = new ByteArrayInputStream(b);
   WorkObjectInputStream wois = null;
   try
   {
      WorkClassLoader wcl = SecurityActions.createWorkClassLoader(classBundle);
      
      wois = new WorkObjectInputStream(bias, wcl);

      DistributableWork dw = (DistributableWork)wois.readObject();

      localDoWork(logicalAddress, dw);
   }
   catch (WorkException we)
   {
      throw we;
   }
   catch (Throwable t)
   {
      throw new WorkException("Error during doWork: " + t.getMessage(), t);
   }
   finally
   {
      if (wois != null)
      {
         try
         {
            wois.close();
         }
         catch (IOException ioe)
         {
            // Ignore
         }
      }
   }
}
 

/**
 * Do first checks for work starting methods
 * @param work to check
 * @param startTimeout to check
 * @param execContext to check
 * @throws WorkException in case of check don't pass
 */
public void doFirstChecks(Work work, long startTimeout, ExecutionContext execContext) throws WorkException
{
   if (isShutdown())
      throw new WorkRejectedException(bundle.workmanagerShutdown());

   if (work == null)
      throw new WorkRejectedException(bundle.workIsNull());

   if (startTimeout < 0)
      throw new WorkRejectedException(bundle.startTimeoutIsNegative(startTimeout));

   checkAndVerifyWork(work, execContext);
}
 

@Override
public void doWork(Work work, long startTimeout, ExecutionContext executionContext, WorkListener workListener)
		throws WorkException {

	Assert.state(this.syncTaskExecutor != null, "No 'syncTaskExecutor' set");
	executeWork(this.syncTaskExecutor, work, startTimeout, false, executionContext, workListener);
}