import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public Object call() throws Exception {
    try {
        long total = 0;
        for (int loop = 0; loop < numRepeats; loop++) {
            // Insert event into named window
            sendMarketBean(threadKey, loop + 1);
            total += loop + 1;

            // iterate over private statement
            SafeIterator safeIter = statement.safeIterator();
            EventBean[] received = EPAssertionUtil.iteratorToArray(safeIter);
            safeIter.close();

            for (int i = 0; i < received.length; i++) {
                if (received[i].get("theString").equals(threadKey)) {
                    long sum = (Long) received[i].get("sumLong");
                    Assert.assertEquals(total, sum);
                }
            }
        }
    } catch (Exception ex) {
        log.error("Error in thread " + Thread.currentThread().getId(), ex);
        return false;
    }
    return true;
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public <T> List<T> concurrentSafeQuery(RowMapper<T> rm) {
	LOG.info("Concurrent safe query being executed");

	if (epStatement.isStopped() || epStatement.isDestroyed()) {
		LOG
				.error("Concurrent safe query was attempted when the statement was stopped or destroyed");
		throw new EsperStatementInvalidStateException(
				"Attempted to execute a concurrent safe query when esper statement resource had state of "
						+ epStatement.getState());
	}

	SafeIterator<EventBean> safeIter = this.epStatement.safeIterator();

	List<T> objectList = new ArrayList<T>();
	try {
		for (; safeIter.hasNext();) {
			EventBean event = safeIter.next();
			objectList.add(rm.mapRow(event));
		}
	} finally {
		safeIter.close();
	}

	LOG.info("Concurrent safe query was completed");
	return objectList;
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
protected List<Object> handleGetAllEvents(String strategyName, String ruleName) {

		EPStatement statement = getServiceProvider(strategyName).getEPAdministrator().getStatement(ruleName);
		List<Object> list = new ArrayList<Object>();
		if (statement != null && statement.isStarted()) {
			SafeIterator<EventBean> it = statement.safeIterator();
			try {
				while (it.hasNext()) {
					EventBean bean = it.next();
					Object underlaying = bean.getUnderlying();
					list.add(underlaying);
				}
			} finally {
				it.close();
			}
		}
		return list;
	}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
protected List<Object> handleGetAllEventsProperty(String strategyName, String ruleName, String property) throws Exception {

		EPStatement statement = getServiceProvider(strategyName).getEPAdministrator().getStatement(ruleName);
		List<Object> list = new ArrayList<Object>();
		if (statement != null && statement.isStarted()) {
			SafeIterator<EventBean> it = statement.safeIterator();
			try {
				while (it.hasNext()) {
					EventBean bean = it.next();
					Object underlaying = bean.get(property);
					list.add(underlaying);
				}
			} finally {
				it.close();
			}
		}
		return list;
	}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
/**
 * This method is used to query a certain statement from the CEP Engine. The method will return an ArrayList of EventBeans
 * @param StatementName
 * @return ArrayList<EventBean> 
 */
private ArrayList<EventBean> queryStatement(String StatementName) {
	
	long start = System.currentTimeMillis();
	
	// Get the Eventbean iterator
	SafeIterator<EventBean> iterator = getEsperStatement(StatementName).safeIterator();
	ArrayList<EventBean> result = new ArrayList<EventBean>();
	logger.trace("Querying statement: " + StatementName);
	
	
	// Loop over the list and add them to the result list.
	while (iterator.hasNext()) {
		EventBean event = iterator.next();
		result.add(event);
	}

	iterator.close();

	long end = System.currentTimeMillis();
	logger.trace("queryStatement took " + (end - start) + " msecs to process " + StatementName );

	
	// Return result list
	return result;
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public Object call() throws Exception {
    try {
        for (int loop = 0; loop < numRepeats; loop++) {
            log.info(".call Thread " + Thread.currentThread().getId() + " sending event " + loop);
            String id = Long.toString(threadNum * 100000000 + loop);
            SupportBean bean = new SupportBean(id, 0);
            engine.getEPRuntime().sendEvent(bean);

            for (int i = 0; i < stmt.length; i++) {
                log.info(".call Thread " + Thread.currentThread().getId() + " starting iterator " + loop);
                SafeIterator<EventBean> it = stmt[i].safeIterator();
                boolean found = false;
                for (; it.hasNext(); ) {
                    EventBean theEvent = it.next();
                    if (theEvent.get("theString").equals(id)) {
                        found = true;
                    }
                }
                it.close();
                Assert.assertTrue(found);
                log.info(".call Thread " + Thread.currentThread().getId() + " end iterator " + loop);
            }
        }
    } catch (AssertionFailedError ex) {
        log.error("Assertion error in thread " + Thread.currentThread().getId(), ex);
        return false;
    } catch (Throwable t) {
        log.error("Error in thread " + Thread.currentThread().getId(), t);
        return false;
    }
    return true;
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public <T> T concurrentSafeQueryForObject(RowMapper<T> rm) {
	LOG.info("Concurrent safe query for object being executed");

	if (epStatement.isStopped() || epStatement.isDestroyed()) {
		LOG
				.error("Concurrent safe query for object was attempted when the statement was stopped or destroyed");
		throw new EsperStatementInvalidStateException(
				"Attempted to execute a concurrent safe query for object when esper statement resource had state of "
						+ epStatement.getState());
	}

	SafeIterator<EventBean> safeIter = this.epStatement.safeIterator();

	T result = null;
	try {
		// Only retrieve the last result
		while (safeIter.hasNext()) {
			EventBean event = safeIter.next();
			if (!safeIter.hasNext()) {
				result = rm.mapRow(event);
			}

		}
	} finally {
		safeIter.close();
	}

	LOG.info("Concurrent safe query for object was completed");
	return result;
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
protected Object handleGetLastEvent(String strategyName, String ruleName) {

		EPStatement statement = getServiceProvider(strategyName).getEPAdministrator().getStatement(ruleName);
		if (statement != null && statement.isStarted()) {
			SafeIterator<EventBean> it = statement.safeIterator();
			try {
				if (it.hasNext()) {
					return it.next().getUnderlying();
				}
			} finally {
				it.close();
			}
		}
		return null;
	}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
protected Object handleGetLastEventProperty(String strategyName, String ruleName, String property) throws Exception {

		EPStatement statement = getServiceProvider(strategyName).getEPAdministrator().getStatement(ruleName);
		if (statement != null && statement.isStarted()) {
			SafeIterator<EventBean> it = statement.safeIterator();
			try {
				return it.next().get(property);
			} finally {
				it.close();
			}
		}
		return null;
	}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public Object call() throws Exception
{
    try
    {
        long total = 0;
        for (int loop = 0; loop < numRepeats; loop++)
        {
            // Insert event into named window
            sendMarketBean(threadKey, loop + 1);
            total += loop + 1;

            // iterate over private statement
            SafeIterator safeIter = statement.safeIterator();
            EventBean[] received = EPAssertionUtil.iteratorToArray(safeIter);
            safeIter.close();

            for (int i = 0; i < received.length; i++)
            {
                if (received[i].get("theString").equals(threadKey))
                {
                    long sum = (Long) received[i].get("sumLong");
                    Assert.assertEquals(total, sum);
                }
            }
        }
    }
    catch (Exception ex)
    {
        log.fatal("Error in thread " + Thread.currentThread().getId(), ex);
        return false;
    }
    return true;
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
private void runAssertionStartStop(EPServiceProvider epService) {
    String epl = "@IterableUnbound every tag=" + SupportBean.class.getName();
    EPStatement patternStmt = epService.getEPAdministrator().createPattern(epl, "MyPattern");
    assertEquals(StatementType.PATTERN, ((EPStatementSPI) patternStmt).getStatementMetadata().getStatementType());

    // Pattern started when created
    assertFalse(patternStmt.iterator().hasNext());
    SafeIterator<EventBean> safe = patternStmt.safeIterator();
    assertFalse(safe.hasNext());
    safe.close();

    // Stop pattern
    patternStmt.stop();
    sendEvent(epService);
    assertNull(patternStmt.iterator());

    // Start pattern
    patternStmt.start();
    assertFalse(patternStmt.iterator().hasNext());

    // Send event
    SupportBean theEvent = sendEvent(epService);
    assertSame(theEvent, patternStmt.iterator().next().get("tag"));
    safe = patternStmt.safeIterator();
    assertSame(theEvent, safe.next().get("tag"));
    safe.close();

    // Stop pattern
    patternStmt.stop();
    assertNull(patternStmt.iterator());

    // Start again, iterator is zero
    patternStmt.start();
    assertFalse(patternStmt.iterator().hasNext());

    // assert statement-eventtype reference info
    EPServiceProviderSPI spi = (EPServiceProviderSPI) epService;
    assertTrue(spi.getStatementEventTypeRef().isInUse(SupportBean.class.getName()));
    Set<String> stmtNames = spi.getStatementEventTypeRef().getStatementNamesForType(SupportBean.class.getName());
    assertTrue(stmtNames.contains("MyPattern"));

    patternStmt.destroy();

    assertFalse(spi.getStatementEventTypeRef().isInUse(SupportBean.class.getName()));
    stmtNames = spi.getStatementEventTypeRef().getStatementNamesForType(SupportBean.class.getName());
    assertFalse(stmtNames.contains("MyPattern"));
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public synchronized SafeIterator<EventBean> safeIterator(int statementId, ContextPartitionSelector selector) {
    AgentInstance[] instances = getAgentInstancesForStmt(statementId, selector);
    return new AgentInstanceArraySafeIterator(instances);
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public synchronized SafeIterator<EventBean> safeIterator(int statementId) {
    AgentInstance[] instances = getAgentInstancesForStmt(statementId);
    return new AgentInstanceArraySafeIterator(instances);
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public SafeIterator<EventBean> safeIterator(int statementId) {
    return iteratorHandler.safeIterator(statementId);
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public synchronized SafeIterator<EventBean> safeIterator(String statementId, ContextPartitionSelector selector) {
    AgentInstance[] instances = getAgentInstancesForStmt(statementId, selector);
    return new AgentInstanceArraySafeIterator(instances);
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public synchronized SafeIterator<EventBean> safeIterator(String statementId) {
    AgentInstance[] instances = getAgentInstancesForStmt(statementId);
    return new AgentInstanceArraySafeIterator(instances);
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public SafeIterator<EventBean> safeIterator(String statementId) {
    return iteratorHandler.safeIterator(statementId);
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
public Object call() throws Exception
{
    try
    {
        for (int loop = 0; loop < numRepeats; loop++)
        {
            log.info(".call Thread " + Thread.currentThread().getId() + " sending event " + loop);
            String id = Long.toString(threadNum * 100000000 + loop);
            SupportBean bean = new SupportBean(id, 0);
            engine.getEPRuntime().sendEvent(bean);

            for (int i = 0; i < stmt.length; i++)
            {
                log.info(".call Thread " + Thread.currentThread().getId() + " starting iterator " + loop);
                SafeIterator<EventBean> it = stmt[i].safeIterator();
                boolean found = false;
                for (;it.hasNext();)
                {
                    EventBean theEvent = it.next();
                    if (theEvent.get("theString").equals(id))
                    {
                        found = true;
                    }
                }
                it.close();
                Assert.assertTrue(found);
                log.info(".call Thread " + Thread.currentThread().getId() + " end iterator " + loop);
            }
        }
    }
    catch (AssertionFailedError ex)
    {
        log.fatal("Assertion error in thread " + Thread.currentThread().getId(), ex);
        return false;
    }
    catch (Throwable t)
    {
        log.fatal("Error in thread " + Thread.currentThread().getId(), t);
        return false;
    }
    return true;
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
/**
 * Compare the property values returned by events of both iterators with the expected values, using exact-order semantics.
 *
 * @param iterator      provides events
 * @param safeIterator  provides events
 * @param propertyNames array of property names
 * @param expected      expected values
 */
public static void assertPropsPerRow(Iterator<EventBean> iterator, SafeIterator<EventBean> safeIterator, String[] propertyNames, Object[][] expected) {
    assertPropsPerRow(EPAssertionUtil.iteratorToArray(iterator), propertyNames, expected);
    assertPropsPerRow(EPAssertionUtil.iteratorToArray(safeIterator), propertyNames, expected);
    safeIterator.close();
}
 

import com.espertech.esper.client.SafeIterator; //导入依赖的package包/类
/**
 * Compare the property values returned by events of both iterators with the expected values, using any-order semantics.
 *
 * @param iterator      provides events
 * @param safeIterator  provides events
 * @param propertyNames array of property names
 * @param expected      expected values
 */
public static void assertPropsPerRowAnyOrder(Iterator<EventBean> iterator, SafeIterator<EventBean> safeIterator, String[] propertyNames, Object[][] expected) {
    assertPropsPerRowAnyOrder(EPAssertionUtil.iteratorToArray(iterator), propertyNames, expected);
    assertPropsPerRowAnyOrder(EPAssertionUtil.iteratorToArray(safeIterator), propertyNames, expected);
    safeIterator.close();
}