SpaceReceiveAdapter extends ReceiverAdapter implements InitializingBean, DisposableBean { @Override public void receive(final Message msg) { final byte[] data = msg.getBuffer(); final ObjectBuffer objectBuffer = new ObjectBuffer( jSpace.getSpaceConfiguration().getKryo() ); final Address nodeRaised = msg.getSrc(); final MethodCall methodCall = (MethodCall) objectBuffer.readClassAndObject( data ); final short id = methodCall.getMethodId(); logger.debug( "received {} from {}", methodCall, nodeRaised ); if ( id == SpaceMethodsMapping.BEGIN_TRANSACTION.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { BeginTransactionMethodCall beginTransactionMethodCall = (BeginTransactionMethodCall) methodCall; long transactionTimeout = beginTransactionMethodCall.getTransactionTimeout(); SpaceTransactionHolder spaceTransactionHolder = new SpaceTransactionHolder(); spaceTransactionHolder.setSynchronizedWithTransaction( true ); spaceTransactionHolder.setTimeoutInMillis( transactionTimeout ); TransactionModificationContext mc = new TransactionModificationContext(); mc.setProxyMode( true ); spaceTransactionHolder.setModificationContext( mc ); modificationContextFor( nodeRaised ).put( mc.getTransactionId(), spaceTransactionHolder ); methodCall.setResponseBody( objectBuffer.writeObjectData( mc.getTransactionId() ) ); } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.COMMIT_TRANSACTION.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { CommitRollbackMethodCall commitMethodCall = (CommitRollbackMethodCall) methodCall; try { SpaceTransactionHolder th = modificationContextFor( nodeRaised ).getIfPresent( commitMethodCall.getTransactionId() ); Preconditions.checkState( th != null, "unable to find transaction with id = %s for commit", commitMethodCall.getTransactionId() ); jSpace.syncTx( th.getModificationContext(), true ); } finally { durableTransactions.remove( commitMethodCall.getTransactionId() ); } } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.ROLLBACK_TRANSACTION.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { CommitRollbackMethodCall commitMethodCall = (CommitRollbackMethodCall) methodCall; try { SpaceTransactionHolder th = modificationContextFor( nodeRaised ).getIfPresent( commitMethodCall.getTransactionId() ); Preconditions.checkState( th != null, "unable to find transaction with id = %s for rollback", commitMethodCall.getTransactionId() ); jSpace.syncTx( th.getModificationContext(), false ); } finally { durableTransactions.remove( commitMethodCall.getTransactionId() ); } } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.WRITE.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { WriteMethodCall writeMethodCall = (WriteMethodCall) methodCall; byte[] entityAndClassData = writeMethodCall.getEntity(); ByteBuffer byteBuffer = ByteBuffer.wrap( entityAndClassData ); int modifiers = writeMethodCall.getModifiers(); int timeout = writeMethodCall.getTimeout(); int timeToLive = writeMethodCall.getTimeToLive(); RegisteredClass entryClass = jSpace.getSpaceConfiguration().getKryo().readClass( byteBuffer ); Class<?> entryType = entryClass.getType(); byte[] entityData = Arrays.copyOfRange( byteBuffer.array(), byteBuffer.position(), byteBuffer.capacity() ); Object entry = jSpace.getSpaceConfiguration().getKryo().readObjectData( byteBuffer, entryType ); SpaceTransactionHolder holder = null; if ( writeMethodCall.getTransactionId() != 0 ) holder = modificationContextFor( nodeRaised ).getIfPresent( writeMethodCall.getTransactionId() ); jSpace.write( holder, entry, entityData, timeToLive, timeout, modifiers ); writeMethodCall.reset(); } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.FETCH.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { FetchMethodCall fetchMethodCall = (FetchMethodCall) methodCall; byte[] entityData = fetchMethodCall.getEntity(); int originalModifiers = fetchMethodCall.getModifiers(); int timeout = fetchMethodCall.getTimeout(); int maxResults = fetchMethodCall.getMaxResults(); Object template = objectBuffer.readClassAndObject( entityData ); int modifiers = originalModifiers | JSpace.RETURN_AS_BYTES; SpaceTransactionHolder holder = null; if ( fetchMethodCall.getTransactionId() != 0 ) holder = modificationContextFor( nodeRaised ).getIfPresent( fetchMethodCall.getTransactionId() ); ByteBuffer[] buffers = (ByteBuffer[]) jSpace.fetch( holder, template, timeout, maxResults, modifiers ); if ( buffers != null ) { byte[][] response = new byte[buffers.length][]; for ( int i = 0; i < buffers.length; i++ ) { ByteBuffer buffer = buffers[i]; response[i] = buffer.array(); } fetchMethodCall.setResponseBody( objectBuffer.writeObjectData( response ) ); } fetchMethodCall.reset(); } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.NOTIFY.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { NotifyListenerMethodCall registerMethodCall = (NotifyListenerMethodCall) methodCall; byte[] entityData = registerMethodCall.getEntity(); int originalModifiers = registerMethodCall.getModifiers(); Object template = objectBuffer.readClassAndObject( entityData ); int modifiers = originalModifiers | JSpace.RETURN_AS_BYTES; jSpace.notify( template, new SpaceNotificationListener() { @Override public void handleNotification(final Object entity, final SpaceOperation operation) { ObjectBuffer innerObjectBuffer = new ObjectBuffer( jSpace.getSpaceConfiguration().getKryo() ); sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { NotifyListenerMethodCall methodCall = new NotifyListenerMethodCall(); methodCall.setEntity( ( (ByteBuffer) entity ).array() ); methodCall.setOperation( operation ); } }, nodeRaised, innerObjectBuffer ); } }, modifiers ); } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.SIZE.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { methodCall.setResponseBody( objectBuffer.writeObjectData( jSpace.size() ) ); } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.MB_USED.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { methodCall.setResponseBody( objectBuffer.writeObjectData( jSpace.mbUsed() ) ); } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.EVICT_ELEMENTS.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { EvictElementsMethodCall evictElementsMethodCall = (EvictElementsMethodCall) methodCall; methodCall.setResponseBody( objectBuffer.writeObjectData( jSpace.evictElements( evictElementsMethodCall.getElements() ) ) ); } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.EVICT_PERCENTAGE.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { EvictPercentageMethodCall evictPercentageMethodCall = (EvictPercentageMethodCall) methodCall; methodCall.setResponseBody( objectBuffer.writeObjectData( jSpace.evictPercentage( evictPercentageMethodCall.getPercentage() ) ) ); } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.EVICT_ALL.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { methodCall.setResponseBody( objectBuffer.writeObjectData( jSpace.evictAll() ) ); } }, nodeRaised, objectBuffer ); else if ( id == SpaceMethodsMapping.SPACE_TOPOLOGY.ordinal() ) sendResponseBackAfterExecution( methodCall, new Runnable() { @Override public void run() { methodCall.setResponseBody( objectBuffer.writeObjectData( jSpace.getSpaceConfiguration().getTopology() ) ); } }, nodeRaised, objectBuffer ); } SpaceReceiveAdapter(final AbstractJSpace jSpace); @Override void afterPropertiesSet(); @Override void destroy(); @Override void receive(final Message msg); @Override void suspect(final Address mbr); @Override void viewAccepted(final View view);  }
@Test public void canPropagateExceptionToClientCode() { CommitRollbackMethodCall commitRollbackMethodCall = new CommitRollbackMethodCall( false ); commitRollbackMethodCall.setTransactionId( System.currentTimeMillis() ); Message message = new Message(); message.setSrc( address ); message.setBuffer( objectBuffer.writeClassAndObject( commitRollbackMethodCall ) ); try { receiveAdapter.receive( message ); } catch ( RuntimeException e ) {} String exceptionAsString = response.getExceptionAsString(); assertThat( exceptionAsString, is( notNullValue() ) ); commitRollbackMethodCall = new CommitRollbackMethodCall( true ); commitRollbackMethodCall.setTransactionId( System.currentTimeMillis() ); message.setBuffer( objectBuffer.writeClassAndObject( commitRollbackMethodCall ) ); try { receiveAdapter.receive( message ); } catch ( RuntimeException e ) {} exceptionAsString = response.getExceptionAsString(); assertThat( exceptionAsString, is( notNullValue() ) ); }
@Test public void canBeginRollback() { BeginTransactionMethodCall methodCall = new BeginTransactionMethodCall(); methodCall.setTransactionTimeout( TimeUnit.SECONDS.toMillis( 1 ) ); Message message = new Message(); message.setSrc( address ); message.setBuffer( objectBuffer.writeClassAndObject( methodCall ) ); receiveAdapter.receive( message ); Long transactionID = objectBuffer.readObjectData( response.getResponseBody(), long.class ); CommitRollbackMethodCall commitRollbackMethodCall = new CommitRollbackMethodCall( false ); commitRollbackMethodCall.setTransactionId( transactionID ); message.setBuffer( objectBuffer.writeClassAndObject( commitRollbackMethodCall ) ); receiveAdapter.receive( message ); String exceptionAsString = response.getExceptionAsString(); assertThat( exceptionAsString, is( nullValue() ) ); }