import org.openspaces.core.GigaSpace; //导入依赖的package包/类
public void testSimpleConfigurer() throws Exception {
    UrlSpaceConfigurer urlSpaceConfigurerPrimary = new UrlSpaceConfigurer("/./space").lookupGroups(System.getProperty("user.name"));

    final AtomicBoolean eventCalled = new AtomicBoolean();

    GigaSpace gigaSpace = new GigaSpaceConfigurer(urlSpaceConfigurerPrimary.space()).gigaSpace();

    SimpleNotifyEventListenerContainer notifyEventListenerContainer = new SimpleNotifyContainerConfigurer(gigaSpace)
            .template(new TestMessage())
            .eventListenerAnnotation(new Object() {
                @SpaceDataEvent
                public void gotMeselfAnEvent() {
                    eventCalled.set(true);
                }
            }).notifyContainer();
    gigaSpace.write(new TestMessage("test"));
    Thread.sleep(200);
    assertTrue(eventCalled.get());

    notifyEventListenerContainer.destroy();
    urlSpaceConfigurerPrimary.destroy();
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
public static <T> AsyncFutureListener<T> wrapIfNeeded(AsyncFutureListener<T> listener, GigaSpace gigaSpace) {
    DefaultTransactionProvider txProvider = (DefaultTransactionProvider) gigaSpace.getTxProvider();
    JiniTransactionHolder holder = txProvider.getHolder();
    PlatformTransactionManager transactionManager = txProvider.getTransactionManager();
    if (holder == null || transactionManager == null) {
        // just wrap for exception translation
        return new InternalAsyncFutureListener<T>(gigaSpace, listener, null, transactionManager, holder);
    } 
    // here, we create a dummy transaction status (with its new transaction set to true, so the commit/roolback
    // process will be performed). We also increase the ref count of the transaction, so only the last one will
    // be performed
    AbstractJiniTransactionManager.JiniTransactionObject jiniTransactionObject = new AbstractJiniTransactionManager.JiniTransactionObject();
    jiniTransactionObject.setJiniHolder(holder, false);
    TransactionStatus txStatus = new DefaultTransactionStatus(jiniTransactionObject, true, false, false, false, null);
    holder.incRef();
    return new InternalAsyncFutureListener<T>(gigaSpace, listener, txStatus, transactionManager, holder);
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    if (args.length == 0) {
        throw new IllegalArgumentException("Must specify the space url");
    }

    String spaceUrl = args[0];
    UrlSpaceConfigurer urlSpaceConfigurer = new UrlSpaceConfigurer(spaceUrl);
    GigaSpace gigaSpace = new GigaSpaceConfigurer(urlSpaceConfigurer.space()).gigaSpace();

    ExecutorSpaceRemotingProxyFactoryBean remotingProxyFactoryBean = new ExecutorSpaceRemotingProxyFactoryBean();
    remotingProxyFactoryBean.setGigaSpace(gigaSpace);
    remotingProxyFactoryBean.setServiceInterface(RefreshableContextLoader.class);
    remotingProxyFactoryBean.setBroadcast(true);
    remotingProxyFactoryBean.afterPropertiesSet();

    RefreshableContextLoader refreshableContextLoader = (RefreshableContextLoader) remotingProxyFactoryBean.getObject();
    refreshableContextLoader.refresh();

    urlSpaceConfigurer.destroy();
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
protected GigaSpace findGigaSpaceByName(String gsName) throws NoSuchBeanDefinitionException {

        initMapsIfNecessary();
        if (gsName == null || "".equals(gsName)) {
            if (this.uniqueGs != null) {
                return this.uniqueGs;
            } else {
                throw new NoSuchBeanDefinitionException("No GigaSpaces name given and factory contains several");
            }
        }
        GigaSpace namedGs = this.gsByName.get(gsName);
        if (namedGs == null) {
            throw new NoSuchBeanDefinitionException("No GigaSpaces found for name [" + gsName + "]");
        }
        return namedGs;
    }
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
private void test(MockArchiveOperationsHandler archiveHandler, GigaSpace gigaSpace, ArchivePollingContainer container, int expectedBatchSize) throws InterruptedException {
      boolean atomic = archiveHandler.supportsBatchArchiving();
      int batchSize = container.getBatchSize();
      int actualBatchSize;
        if (atomic) {
            actualBatchSize = batchSize;
        }
        else {
            Assert.assertEquals("Test configuration error. Cannot expect batchSize!=1 if not atomic, since the implementation uses take and not takeMultiple", 1, expectedBatchSize);
            actualBatchSize = 1;
        }
        
        Assert.assertEquals(expectedBatchSize, actualBatchSize);
        Assert.assertTrue("Atomic test must be performed with a space that uses a transaction manager", !atomic || gigaSpace.getTxProvider() != null);
        Assert.assertTrue("Atomic test must be performed with a container that uses a transaction manager", !atomic || container.getTransactionManagerName() != null);

        test(archiveHandler, gigaSpace, actualBatchSize);
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
/**
 * Performs the receive operation. If blocking, we call {@link #doReceiveBlocking(Object,org.openspaces.core.GigaSpace,long)}
 * and expect it to block for the receive timeout (or until a match is found). If non blocking, will perform
 * {@link #setNonBlockingFactor(int)} number of {@link #doReceiveNonBlocking(Object,org.openspaces.core.GigaSpace)}
 * operations within the receive timeout (default factor is 10).
 */
public Object receive(Object template, GigaSpace gigaSpace, long receiveTimeout) throws DataAccessException {
    if (!nonBlocking) {
        return doReceiveBlocking(template, gigaSpace, receiveTimeout);
    }
    long sleepTime = receiveTimeout / nonBlockingFactor;
    for (int i = 0; i < nonBlockingFactor; i++) {
        Object event = doReceiveNonBlocking(template, gigaSpace);
        if (event != null) {
            return event;
        }
        try {
            Thread.sleep(sleepTime);
        } catch (InterruptedException e) {
            throw new SpaceInterruptedException("Interrupted while performing non blocking receive operation");
        }
    }
    return null;
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
public AnnotatedMember(String name, AccessibleObject member) {
    this.name = name;
    this.member = member;

    // Validate member type
    Class<?> memberType = getMemberType();
    if (!GigaSpace.class.isAssignableFrom(memberType)) {
        throw new IllegalArgumentException("Cannot inject [" + member + "], not a supported GigaSpaces type");
    }
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
@Bean
public GigaSpace otherEmbeddedGigaSpace() {
	IJSpace space = Mockito.mock(IJSpace.class);
	GigaSpace gs = Mockito.mock(GigaSpace.class);
	Mockito.stub(gs.getSpace()).toReturn(space);
	Mockito.stub(space.isEmbedded()).toReturn(true);
	return gs;
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
public MachinesStateBackupToSpace(Admin admin, GigaSpace space, MachinesSlaEnforcementState machinesSlaEnforcementState) {
    this.admin = (InternalAdmin) admin;
    this.space = space;
    this.machinesSlaEnforcementState = machinesSlaEnforcementState;
    writeCompletedVersion = new AtomicLong(machinesSlaEnforcementState.getVersion() - 1);
    lastError = new AtomicReference<Throwable>();
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
public GigaSpace getGigaSpace() {
    if (gigaSpace == null) {
        
        IJSpace ijSpace = getIJSpace();
        if (ijSpace == null) {
            throw new AdminException("Cannot create GigaSpace object since no space instance is discovered");
        }
        this.gigaSpace = new GigaSpaceConfigurer(ijSpace).clustered(true).gigaSpace();
    }
    return this.gigaSpace;
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
private synchronized void initializeScriptingExecutorProxy() {
    if (_scriptingExecutor == null) {
        boolean clustered = ((ISpaceProxy) getSpace()).isClustered();
        GigaSpace gigaSpace = new GigaSpaceConfigurer(getSpace()).clustered(clustered).gigaSpace();
        _scriptingExecutor =  new ExecutorRemotingProxyConfigurer<ScriptingExecutor>(gigaSpace, ScriptingExecutor.class).proxy();
    }        
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
/**
 * Lazily initialize gs map.
 */
private synchronized void initMapsIfNecessary() {
    if (this.gsByName == null) {
        this.gsByName = new HashMap<String, GigaSpace>();
        // Look for named GigaSpaces

        for (String gsName : BeanFactoryUtils.beanNamesForTypeIncludingAncestors(this.applicationContext, GigaSpace.class)) {

            GigaSpace gs = (GigaSpace) this.applicationContext.getBean(gsName);
            gsByName.put(gsName, gs);
        }

        if (this.gsByName.isEmpty()) {
            // Try to find a unique GigaSpaces.
            String[] gsNames = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(this.applicationContext, GigaSpace.class);
            if (gsNames.length == 1) {
                this.uniqueGs = (GigaSpace) this.applicationContext.getBean(gsNames[0]);
            }
        } else if (this.gsByName.size() == 1) {
            this.uniqueGs = this.gsByName.values().iterator().next();
        }

        if (this.gsByName.isEmpty() && this.uniqueGs == null) {
            logger.warn("No named gs instances defined and not exactly one anonymous one: cannot inject");
        }
    }
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
/**
 * Creates a new <code>ExecutorScriptingProxyConfigurer</code> on top of the given space
 */
public ExecutorScriptingProxyConfigurer(GigaSpace gigaSpace) {
    remotingConfigurer = new ExecutorRemotingProxyConfigurer<ScriptingExecutor>(gigaSpace, ScriptingExecutor.class)
            .metaArgumentsHandler(new ScriptingMetaArgumentsHandler())
            .remoteInvocationAspect(new LazyLoadingRemoteInvocationAspect())
            .remoteRoutingHandler(new ScriptingRemoteRoutingHandler());
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
private void writeResponse(GigaSpace gigaSpace, SpaceRemotingEntry remotingEntry, Throwable e) {
    if (remotingEntry.getOneWay() == null || !remotingEntry.getOneWay()) {
        SpaceRemotingEntry result = remotingEntry.buildResult(e);
        if (clusterInfo != null) {
            result.setInstanceId(clusterInfo.getInstanceId());
        }
        gigaSpace.write(result);
    } else {
        if (logger.isDebugEnabled()) {
            logger.debug("Remoting execution is configured as one way and an exception was thrown", e);
        }
    }
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
@Test(expected=ChangeException.class)
public void testAddAndGetIllegalBroadcastResult() {
    GigaSpace gigaSpace = mock(GigaSpace.class);
    int delta = 2;
    String path = "path";
    IdQuery<ChangeExtensionPojo> query = new IdQuery<ChangeExtensionPojo>(ChangeExtensionPojo.class, "id");
    
    ChangeResult changeResult = mock(ChangeResult.class);
    when(changeResult.getNumberOfChangedEntries()).thenReturn(2);
    when(gigaSpace.change(any(ISpaceQuery.class), any(ChangeSet.class), any(ChangeModifiers.class), anyLong())).thenReturn(changeResult);
    
    ChangeExtension.addAndGet(gigaSpace, query, path, delta);
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
@Test
public void itsPossibleToInvokeMethodsReturningAsyncFutureWithFaultTolerance() throws Exception {
	configurer.setSubsystem("lunch-system");
	configurer.enableFaultTolerance(true);
	astrix = autoClosables.add(configurer.configure());
	GigaSpace gigaSpace = astrix.waitForBean(GigaSpace.class, "lunch-space", 10000);
	AsyncFuture<LunchRestaurant> future = gigaSpace.asyncRead(LunchRestaurant.template());
	assertEquals(AsyncFutureImpl.class, future.getClass());
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
@Test(expected=IllegalArgumentException.class)
public void testAddAndGetNullDeltaThrowException() {
    GigaSpace gigaSpace = mock(GigaSpace.class);
    Integer delta = null;
    String path = "path";
    IdQuery<ChangeExtensionPojo> query = new IdQuery<ChangeExtensionPojo>(ChangeExtensionPojo.class, "id");
    ChangeExtension.addAndGet(gigaSpace, query, path, delta);
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
@Test
public void testSimpleTaskExecution() throws Exception {
    GigaSpace gigaSpace = (GigaSpace) applicationContext.getBean("clusteredGigaSpace");
    AsyncFuture<Long> future = gigaSpace.execute(new MyTask());
    long result = future.get();
    Assert.assertEquals(1, result);
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
@Test
public void testAddAndGetTimeout() {
    GigaSpace gigaSpace = mock(GigaSpace.class);
    int delta = 2;
    String path = "path";
    IdQuery<ChangeExtensionPojo> query = new IdQuery<ChangeExtensionPojo>(ChangeExtensionPojo.class, "id");
    
    ChangeResult changeResult = mock(ChangeResult.class);
    when(gigaSpace.change(any(ISpaceQuery.class), any(ChangeSet.class), any(ChangeModifiers.class), anyLong())).thenReturn(changeResult);
    
    ChangeExtension.addAndGet(gigaSpace, query, path, delta, 1, TimeUnit.SECONDS);        
    verify(gigaSpace).change(eq(query), any(ChangeSet.class), eq(ChangeModifiers.RETURN_DETAILED_RESULTS), eq(1000L));
}
 

import org.openspaces.core.GigaSpace; //导入依赖的package包/类
@Override
public <T> BoundServiceBeanInstance<T> bind(
		ServiceDefinition<T> serviceDefinition,
		ServiceProperties serviceProperties) {
	Class<T> type = serviceDefinition.getServiceType();
	if (!GigaSpace.class.isAssignableFrom(type)) {
		throw new UnsupportedTargetTypeException(getName(), type);
	}
	if (disableLocalView.get()) {
		log.info("LocalView is disabled. Creating reqular proxy");
		return clusteredProxyBinder.bind(serviceDefinition, serviceProperties);
	}
	log.info("Creating local view. bean={} serviceProperties={}", serviceDefinition.getBeanKey(), serviceProperties.getProperties());
	// TODO: protect creation of localView with fault-tolerance?
	Class<LocalViewConfigurer> serviceConfigClass = serviceDefinition.getServiceConfigClass(LocalViewConfigurer.class);	
	LocalViewConfigurer localViewConfigurer = ReflectionUtil.newInstance(serviceConfigClass);
	UrlSpaceConfigurer gsSpaceConfigurer = new UrlSpaceConfigurer(serviceProperties.getProperty(GsBinder.SPACE_URL_PROPERTY));
	IJSpace space = gsSpaceConfigurer.lookupTimeout(this.lookupTimeout.get())
			// Disable memory shortage check for local view clients
			.addParameter("space-config.engine.memory_usage.enabled", "false").create();
	
	LocalViewSpaceConfigurer gslocalViewSpaceConfigurer = new LocalViewSpaceConfigurer(space);
	gslocalViewSpaceConfigurer.maxDisconnectionDuration(this.maxDisonnectionTime.get());
	localViewConfigurer.configure(new LocalViewSpaceConfigurerAdapter(gslocalViewSpaceConfigurer));
	
	String spaceName = serviceProperties.getProperty(GsBinder.SPACE_NAME_PROPERTY);
	String commandKey = spaceName + "_" + GigaSpace.class.getSimpleName();
	String qualifier = serviceProperties.getProperty(ServiceProperties.QUALIFIER);
	if (qualifier != null) {
		commandKey = commandKey + "-" + qualifier;
	}
	
	IJSpace localViewSpace = gslocalViewSpaceConfigurer.create();
	GigaSpace localViewGigaSpace = GigaSpaceProxy.create(new GigaSpaceConfigurer(localViewSpace).create());
	
	BoundLocalViewGigaSpaceBeanInstance localViewGigaSpaceBeanInstance = 
			new BoundLocalViewGigaSpaceBeanInstance(localViewGigaSpace, gslocalViewSpaceConfigurer, gsSpaceConfigurer);
	return (BoundServiceBeanInstance<T>) localViewGigaSpaceBeanInstance;
}