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

/**
 * Test unsupported context nested startWork
 * @throws Throwable throwable exception 
 */
//@Test
public void testStartWorkUnsupportedContext() throws Throwable
{
   ContextWorkAdapter wa = new ContextWorkAdapter();
   NestProviderWork workA = new NestProviderWork("A", wa);
   workA.addContext(new HintsContext());

   NestProviderWork workB = new NestProviderWork("B", null);
   workB.addContext(new UnsupportedContext());

   WorkConnection wc = wcf.getConnection();
   workA.setNestDo(false);
   workA.setWorkManager(wc.getWorkManager());
   workA.setWork(workB);
   
   CyclicBarrier barrier = new CyclicBarrier(2);
   workA.setBarrier(barrier);
   workB.setBarrier(barrier);

   wc.startWork(workA, WorkManager.INDEFINITE, null, wa);
   barrier.await();
   assertNotNull(wa.getException());

   wc.close();
}
 

/**
 * Test for bullet 4 Section 3.3.6
 * When the application server is unable to recreate an execution context if it is  
 *                      specified for the submitted Work instance, it must throw a
 *                      WorkCompletedException set to an appropriate error code.
 * @throws Throwable throwable exception 
 */
@Test(expected = WorkCompletedException.class)
public void testThrowWorkCompletedException() throws Throwable
{
   ExecutionContext ec = new ExecutionContext();
   ShortRunningWork work = new ShortRunningWork();
   ec.setXid(new XidImpl());
   ec.setTransactionTimeout(Long.MAX_VALUE);

   WorkConnection wc = wcf.getConnection();
   try
   {
      wc.doWork(work, WorkManager.INDEFINITE, ec, null);
   }
   finally
   {
      wc.close();
   }
}
 

/**
 * Test for paragraph 3
 * doWork method: this provides a first in, first out (FIFO) execution start 
 *      ordering and last in, first out (LIFO) execution completion ordering guarantee.
 * @throws Throwable throwable exception 
 */
@Test
public void testFifoStartLifoFinish() throws Throwable
{
   ContextWorkAdapter wa = new ContextWorkAdapter();
   NestProviderWork workA = new NestProviderWork("A", wa);
   workA.addContext(new TransactionContext());

   NestProviderWork workB = new NestProviderWork("B", null);
   workB.addContext(new WorkContextSetupListenerTransactionContext());

   WorkConnection wc = wcf.getConnection();
   workA.setNestDo(true);
   workA.setWorkManager(wc.getWorkManager());
   workA.setWork(workB);
   wc.doWork(workA, WorkManager.INDEFINITE, null, wa);

   assertEquals(wa.getStart(), "AB");
   assertEquals(wa.getDone(), "BA");

   wc.close();
}
 

/**
 * Test unsupported context nested doWork. 
 * @throws Throwable throwable exception 
 */
//@Test(expected = Throwable.class)
public void testDoWorkUnsupportedContext() throws Throwable
{
   ContextWorkAdapter wa = new ContextWorkAdapter();
   NestProviderWork workA = new NestProviderWork("A", wa);
   workA.addContext(new TransactionContext());

   NestProviderWork workB = new NestProviderWork("B", null);
   workB.addContext(new UnsupportedContext());

   WorkConnection wc = wcf.getConnection();
   workA.setNestDo(true);
   workA.setWorkManager(wc.getWorkManager());
   workA.setWork(workB);
   wc.doWork(workA, WorkManager.INDEFINITE, null, wa);
   wc.close();
}
 

/**
 * Obtain the default JBoss JCA WorkManager through a JMX lookup
 * for the JBossWorkManagerMBean.
 * @param mbeanName the JMX object name to use
 * @see org.jboss.resource.work.JBossWorkManagerMBean
 */
public static WorkManager getWorkManager(String mbeanName) {
	Assert.hasLength(mbeanName, "JBossWorkManagerMBean name must not be empty");
	try {
		Class<?> mbeanClass = JBossWorkManagerUtils.class.getClassLoader().loadClass(JBOSS_WORK_MANAGER_MBEAN_CLASS_NAME);
		InitialContext jndiContext = new InitialContext();
		MBeanServerConnection mconn = (MBeanServerConnection) jndiContext.lookup(MBEAN_SERVER_CONNECTION_JNDI_NAME);
		ObjectName objectName = ObjectName.getInstance(mbeanName);
		Object workManagerMBean = MBeanServerInvocationHandler.newProxyInstance(mconn, objectName, mbeanClass, false);
		Method getInstanceMethod = workManagerMBean.getClass().getMethod("getInstance");
		return (WorkManager) getInstanceMethod.invoke(workManagerMBean);
	}
	catch (Exception ex) {
		throw new IllegalStateException(
				"Could not initialize JBossWorkManagerTaskExecutor because JBoss API is not available", ex);
	}
}
 

/**
 * run method
 */
public void run()
{
   try
   {
      if (nestWork != null && workManager != null)
      {
         if (nestDo)
            workManager.doWork(nestWork, WorkManager.INDEFINITE, null, wa);
         else
            workManager.startWork(nestWork, WorkManager.INDEFINITE, null, wa);
      }
   }
   catch (Throwable e)
   {
      throw new RuntimeException(e.getMessage());
   }
}
 

/**
 * Clone the WorkManager implementation
 * @return A copy of the implementation
 * @exception CloneNotSupportedException Thrown if the copy operation isn't supported
 *
 */
@Override
public org.jboss.jca.core.api.workmanager.WorkManager clone() throws CloneNotSupportedException
{
   DistributedWorkManagerImpl wm = (DistributedWorkManagerImpl)super.clone();
   wm.listeners = Collections.synchronizedList(new ArrayList<NotificationListener>(3));
   wm.setPolicy(getPolicy());
   wm.setSelector(getSelector());
   wm.setTransport(getTransport());
   wm.setDistributedStatisticsEnabled(isDistributedStatisticsEnabled());
   wm.setDoWorkDistributionEnabled(isDoWorkDistributionEnabled());
   wm.setStartWorkDistributionEnabled(isStartWorkDistributionEnabled());
   wm.setScheduleWorkDistributionEnabled(isScheduleWorkDistributionEnabled());
   
   return wm;
}
 

@Override
protected void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
	beanFactory.addBeanPostProcessor(new BootstrapContextAwareProcessor(this.bootstrapContext));
	beanFactory.ignoreDependencyInterface(BootstrapContextAware.class);
	beanFactory.registerResolvableDependency(BootstrapContext.class, this.bootstrapContext);

	// JCA WorkManager resolved lazily - may not be available.
	beanFactory.registerResolvableDependency(WorkManager.class,
			(ObjectFactory<WorkManager>) this.bootstrapContext::getWorkManager);
}
 

/**
 * {@inheritDoc}
 */
@Override
public void scheduleWork(Work work) throws WorkException
{
   if (policy == null || selector == null || transport == null ||
       work == null || !(work instanceof DistributableWork) || !scheduleWorkDistributionEnabled)
   {
      localScheduleWork(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.scheduleWork(dwmAddress, dw);
            executed = true;
         }
      }

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

/**
 * {@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);
   }
}
 

/**
 * {@inheritDoc}
 */
public void localScheduleWork(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);

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

@Override
protected void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
	beanFactory.addBeanPostProcessor(new BootstrapContextAwareProcessor(this.bootstrapContext));
	beanFactory.ignoreDependencyInterface(BootstrapContextAware.class);
	beanFactory.registerResolvableDependency(BootstrapContext.class, this.bootstrapContext);

	// JCA WorkManager resolved lazily - may not be available.
	beanFactory.registerResolvableDependency(WorkManager.class,
			(ObjectFactory<WorkManager>) this.bootstrapContext::getWorkManager);
}
 

@Override
public void afterPropertiesSet() throws NamingException {
	if (this.workManager == null) {
		if (this.workManagerName != null) {
			this.workManager = lookup(this.workManagerName, WorkManager.class);
		}
		else {
			this.workManager = getDefaultWorkManager();
		}
	}
}
 

@Override
public WorkManager getWorkManager() {
	if (this.workManager == null) {
		throw new IllegalStateException("No WorkManager available");
	}
	return this.workManager;
}
 

/**
 * Test WorkContextLifecycleListener for transaction context.
 *
 * @throws Throwable throws any error
 */
@Test
public void testTransactionContextCustomListener() throws Throwable
{
   UniversalProviderWork work = new UniversalProviderWork();
   WorkContextSetupListenerTransactionContext listener = new WorkContextSetupListenerTransactionContext();
   work.addContext(listener);
   ContextWorkAdapter wa = new ContextWorkAdapter();
   WorkConnection wc = wcf.getConnection();
   wc.doWork(work, WorkManager.INDEFINITE, null, wa);

   assertEquals("", listener.getContextSetupFailedErrorCode());
   assertTrue(listener.isContextSetupComplete());

   LOG.info("1Test//accepted:" + wa.getTimeAccepted() + "//started:" + wa.getTimeStarted() + "//context:"
         + listener.getTimeStamp() + "//completed:" + wa.getTimeCompleted());

   assertTrue(wa.getTimeAccepted() > 0);
   assertTrue(wa.getTimeStarted() > 0);
   assertTrue(listener.getTimeStamp() > 0);
   assertTrue(wa.getTimeCompleted() > 0);

   assertTrue(wa.getTimeAccepted() <= wa.getTimeStarted());
   assertTrue(wa.getTimeStarted() <= listener.getTimeStamp());
   assertTrue(listener.getTimeStamp() <= wa.getTimeCompleted());
   wc.close();
}
 

@Override
public void afterPropertiesSet() throws NamingException {
	if (this.workManager == null) {
		if (this.workManagerName != null) {
			this.workManager = lookup(this.workManagerName, WorkManager.class);
		}
		else {
			this.workManager = getDefaultWorkManager();
		}
	}
}
 

/**
 * Identify a specific GlassFish thread pool to talk to.
 * <p>The thread pool name matches the resource adapter name
 * in default RAR deployment scenarios.
 */
public void setThreadPoolName(String threadPoolName) {
	WorkManager wm = (WorkManager) ReflectionUtils.invokeMethod(this.getWorkManagerMethod, null, threadPoolName);
	if (wm == null) {
		throw new IllegalArgumentException("Specified thread pool name '" + threadPoolName +
				"' does not correspond to an actual pool definition in GlassFish. Check your configuration!");
	}
	setWorkManager(wm);
}
 

/**
 * {@inheritDoc}
 */
public void localDoWork(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);

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

/**
 * {@inheritDoc}
 */
public void localScheduleWork(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);

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

/**
 * {@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);
   }
}
 

/**
 * {@inheritDoc}
 */
@Override
public void scheduleWork(Work work) throws WorkException
{
   if (policy == null || selector == null || transport == null ||
       work == null || !(work instanceof DistributableWork) || !scheduleWorkDistributionEnabled)
   {
      localScheduleWork(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.scheduleWork(dwmAddress, dw);
            executed = true;
         }
      }

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

@Override
protected void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
	beanFactory.addBeanPostProcessor(new BootstrapContextAwareProcessor(this.bootstrapContext));
	beanFactory.ignoreDependencyInterface(BootstrapContextAware.class);
	beanFactory.registerResolvableDependency(BootstrapContext.class, this.bootstrapContext);

	// JCA WorkManager resolved lazily - may not be available.
	beanFactory.registerResolvableDependency(WorkManager.class, new ObjectFactory<WorkManager>() {
		@Override
		public WorkManager getObject() {
			return bootstrapContext.getWorkManager();
		}
	});
}
 

@Override
public void afterPropertiesSet() throws NamingException {
	if (this.workManager == null) {
		if (this.workManagerName != null) {
			this.workManager = lookup(this.workManagerName, WorkManager.class);
		}
		else {
			this.workManager = getDefaultWorkManager();
		}
	}
}
 

/**
 * {@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);
      }
   }
}
 

/**
 * Identify a specific GlassFish thread pool to talk to.
 * <p>The thread pool name matches the resource adapter name
 * in default RAR deployment scenarios.
 */
public void setThreadPoolName(String threadPoolName) {
	WorkManager wm = (WorkManager) ReflectionUtils.invokeMethod(this.getWorkManagerMethod, null, threadPoolName);
	if (wm == null) {
		throw new IllegalArgumentException("Specified thread pool name '" + threadPoolName +
				"' does not correspond to an actual pool definition in GlassFish. Check your configuration!");
	}
	setWorkManager(wm);
}
 

/**
 * Get the work manager
 *
 * @return The manager
 */
public WorkManager getWorkManager() {
   if (logger.isTraceEnabled()) {
      logger.trace("getWorkManager()");
   }

   if (ctx == null) {
      return null;
   }

   return ctx.getWorkManager();
}
 

/**
 * Get the work manager
 *
 * @return The value
 */
public WorkManager getWorkManager() {
   if (logger.isTraceEnabled()) {
      logger.trace("getWorkManager()");
   }

   return ra.getWorkManager();
}
 

public WorkManager getWorkManager() {
    if (getBootstrapContext() instanceof BootstrapContext) {
        BootstrapContext context = (BootstrapContext)getBootstrapContext();
        return context.getWorkManager();
    }
    return null;
}
 

public WorkManager getWorkManager() {
    if (resourceAdapter instanceof ResourceAdapterImpl) {
        return ((ResourceAdapterImpl)resourceAdapter).getBootstrapContext()
            .getWorkManager();
    }
    return null;
}
 

@Override
public WorkManager getWorkManager() {
	Assert.state(this.workManager != null, "No WorkManager available");
	return this.workManager;
}