@SuppressWarnings( "unchecked" ) @Test public void processEntityChanges() { final int schemaId = 1; final ChannelSchema channelSchema = new ChannelSchema( 0, ValueUtil.randomString(), null, ChannelSchema.FilterType.NONE, null, false, true, Collections.emptyList() ); final EntitySchema.Creator<Linkable> creator = mock( EntitySchema.Creator.class ); final EntitySchema.Updater<Linkable> updater = mock( EntitySchema.Updater.class ); final EntitySchema entitySchema = new EntitySchema( 0, ValueUtil.randomString(), Linkable.class, creator, updater ); final SystemSchema schema = new SystemSchema( schemaId, ValueUtil.randomString(), new ChannelSchema[]{ channelSchema }, new EntitySchema[]{ entitySchema } ); final Connector connector = createConnector( schema ); connector.setLinksToProcessPerTick( 1 ); final Connection connection = newConnection( connector ); final Linkable userObject1 = mock( Linkable.class ); final Linkable userObject2 = mock( Linkable.class ); pauseScheduler(); final ChannelAddress address = new ChannelAddress( connector.getSchema().getId(), 1 ); final Subscription subscription = createSubscription( address, null, true ); final Entity entity2 = findOrCreateEntity( Linkable.class, 2 ); safeAction( () -> entity2.setUserObject( userObject2 ) ); safeAction( () -> entity2.linkToSubscription( subscription ) ); final Entity entity3 = findOrCreateEntity( Linkable.class, 3 ); final EntityChangeData data1 = mock( EntityChangeData.class ); final EntityChangeData data2 = mock( EntityChangeData.class ); final EntityChange[] entityChanges = { EntityChange.create( 0, 1, new String[]{ "1" }, data1 ), EntityChange.create( 0, 2, new String[]{ "1" }, data2 ), EntityChange.create( 0, 3, new String[]{ "1" } ) }; final MessageResponse response = setCurrentMessageResponse( connection, ChangeSetMessage.create( null, null, null, null, entityChanges ) ); when( creator.createEntity( 1, data1 ) ).thenReturn( userObject1 ); assertEquals( response.getEntityUpdateCount(), 0 ); assertEquals( response.getEntityRemoveCount(), 0 ); connector.setChangesToProcessPerTick( 1 ); connector.processEntityChanges(); verify( creator, times( 1 ) ).createEntity( 1, data1 ); verify( updater, never() ).updateEntity( userObject1, data1 ); verify( creator, never() ).createEntity( 2, data2 ); verify( updater, never() ).updateEntity( userObject2, data2 ); assertEquals( response.getEntityUpdateCount(), 1 ); assertEquals( response.getEntityRemoveCount(), 0 ); connector.setChangesToProcessPerTick( 2 ); connector.processEntityChanges(); verify( creator, times( 1 ) ).createEntity( 1, data1 ); verify( updater, never() ).updateEntity( userObject1, data1 ); verify( creator, never() ).createEntity( 2, data2 ); verify( updater, times( 1 ) ).updateEntity( userObject2, data2 ); assertEquals( response.getEntityUpdateCount(), 2 ); assertEquals( response.getEntityRemoveCount(), 1 ); assertFalse( Disposable.isDisposed( entity2 ) ); assertTrue( Disposable.isDisposed( entity3 ) ); }
@SuppressWarnings( "unchecked" ) @Action void processEntityChanges() { final MessageResponse response = ensureCurrentMessageResponse(); EntityChange change; for ( int i = 0; i < _changesToProcessPerTick && null != ( change = response.nextEntityChange() ); i++ ) { final String id = change.getId(); final int idSeparator = id.indexOf( "." ); final int typeId = Integer.parseInt( id.substring( 0, idSeparator ) ); final int entityId = Integer.parseInt( id.substring( idSeparator + 1 ) ); final EntitySchema entitySchema = getSchema().getEntity( typeId ); final Class<?> type = entitySchema.getType(); Entity entity = getReplicantContext().getEntityService().findEntityByTypeAndId( type, entityId ); if ( change.isRemove() ) { if ( null != entity ) { Disposable.dispose( entity ); response.incEntityRemoveCount(); } } else { final EntityChangeData data = change.getData(); if ( null == entity ) { final String name = Replicant.areNamesEnabled() ? entitySchema.getName() + "/" + entityId : null; entity = getReplicantContext().getEntityService().findOrCreateEntity( name, type, entityId ); final Object userObject = entitySchema.getCreator().createEntity( entityId, data ); entity.setUserObject( userObject ); } else { @SuppressWarnings( "rawtypes" ) final EntitySchema.Updater updater = entitySchema.getUpdater(); if ( null != updater ) { updater.updateEntity( entity.getUserObject(), data ); } } final String[] channels = change.getChannels(); final int schemaId = getSchema().getId(); for ( final String channel : channels ) { final int separator = channel.indexOf( "." ); final ChannelAddress address = new ChannelAddress( schemaId, Integer.parseInt( -1 == separator ? channel : channel.substring( 0, separator ) ), -1 == separator ? null : Integer.parseInt( channel.substring( separator + 1 ) ) ); final Subscription subscription = getReplicantContext().findSubscription( address ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> null != subscription, () -> "Replicant-0069: ChangeSet contained an EntityChange message referencing channel " + address + " but no such subscription exists locally." ); } assert null != subscription; entity.tryLinkToSubscription( subscription ); } response.incEntityUpdateCount(); response.changeProcessed( entity.getUserObject() ); } } }
Connector extends ReplicantService { @SuppressWarnings( "unchecked" ) @Action void processEntityChanges() { final MessageResponse response = ensureCurrentMessageResponse(); EntityChange change; for ( int i = 0; i < _changesToProcessPerTick && null != ( change = response.nextEntityChange() ); i++ ) { final String id = change.getId(); final int idSeparator = id.indexOf( "." ); final int typeId = Integer.parseInt( id.substring( 0, idSeparator ) ); final int entityId = Integer.parseInt( id.substring( idSeparator + 1 ) ); final EntitySchema entitySchema = getSchema().getEntity( typeId ); final Class<?> type = entitySchema.getType(); Entity entity = getReplicantContext().getEntityService().findEntityByTypeAndId( type, entityId ); if ( change.isRemove() ) { if ( null != entity ) { Disposable.dispose( entity ); response.incEntityRemoveCount(); } } else { final EntityChangeData data = change.getData(); if ( null == entity ) { final String name = Replicant.areNamesEnabled() ? entitySchema.getName() + "/" + entityId : null; entity = getReplicantContext().getEntityService().findOrCreateEntity( name, type, entityId ); final Object userObject = entitySchema.getCreator().createEntity( entityId, data ); entity.setUserObject( userObject ); } else { @SuppressWarnings( "rawtypes" ) final EntitySchema.Updater updater = entitySchema.getUpdater(); if ( null != updater ) { updater.updateEntity( entity.getUserObject(), data ); } } final String[] channels = change.getChannels(); final int schemaId = getSchema().getId(); for ( final String channel : channels ) { final int separator = channel.indexOf( "." ); final ChannelAddress address = new ChannelAddress( schemaId, Integer.parseInt( -1 == separator ? channel : channel.substring( 0, separator ) ), -1 == separator ? null : Integer.parseInt( channel.substring( separator + 1 ) ) ); final Subscription subscription = getReplicantContext().findSubscription( address ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> null != subscription, () -> "Replicant-0069: ChangeSet contained an EntityChange message referencing channel " + address + " but no such subscription exists locally." ); } assert null != subscription; entity.tryLinkToSubscription( subscription ); } response.incEntityUpdateCount(); response.changeProcessed( entity.getUserObject() ); } } } }
Connector extends ReplicantService { @SuppressWarnings( "unchecked" ) @Action void processEntityChanges() { final MessageResponse response = ensureCurrentMessageResponse(); EntityChange change; for ( int i = 0; i < _changesToProcessPerTick && null != ( change = response.nextEntityChange() ); i++ ) { final String id = change.getId(); final int idSeparator = id.indexOf( "." ); final int typeId = Integer.parseInt( id.substring( 0, idSeparator ) ); final int entityId = Integer.parseInt( id.substring( idSeparator + 1 ) ); final EntitySchema entitySchema = getSchema().getEntity( typeId ); final Class<?> type = entitySchema.getType(); Entity entity = getReplicantContext().getEntityService().findEntityByTypeAndId( type, entityId ); if ( change.isRemove() ) { if ( null != entity ) { Disposable.dispose( entity ); response.incEntityRemoveCount(); } } else { final EntityChangeData data = change.getData(); if ( null == entity ) { final String name = Replicant.areNamesEnabled() ? entitySchema.getName() + "/" + entityId : null; entity = getReplicantContext().getEntityService().findOrCreateEntity( name, type, entityId ); final Object userObject = entitySchema.getCreator().createEntity( entityId, data ); entity.setUserObject( userObject ); } else { @SuppressWarnings( "rawtypes" ) final EntitySchema.Updater updater = entitySchema.getUpdater(); if ( null != updater ) { updater.updateEntity( entity.getUserObject(), data ); } } final String[] channels = change.getChannels(); final int schemaId = getSchema().getId(); for ( final String channel : channels ) { final int separator = channel.indexOf( "." ); final ChannelAddress address = new ChannelAddress( schemaId, Integer.parseInt( -1 == separator ? channel : channel.substring( 0, separator ) ), -1 == separator ? null : Integer.parseInt( channel.substring( separator + 1 ) ) ); final Subscription subscription = getReplicantContext().findSubscription( address ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> null != subscription, () -> "Replicant-0069: ChangeSet contained an EntityChange message referencing channel " + address + " but no such subscription exists locally." ); } assert null != subscription; entity.tryLinkToSubscription( subscription ); } response.incEntityUpdateCount(); response.changeProcessed( entity.getUserObject() ); } } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); }
Connector extends ReplicantService { @SuppressWarnings( "unchecked" ) @Action void processEntityChanges() { final MessageResponse response = ensureCurrentMessageResponse(); EntityChange change; for ( int i = 0; i < _changesToProcessPerTick && null != ( change = response.nextEntityChange() ); i++ ) { final String id = change.getId(); final int idSeparator = id.indexOf( "." ); final int typeId = Integer.parseInt( id.substring( 0, idSeparator ) ); final int entityId = Integer.parseInt( id.substring( idSeparator + 1 ) ); final EntitySchema entitySchema = getSchema().getEntity( typeId ); final Class<?> type = entitySchema.getType(); Entity entity = getReplicantContext().getEntityService().findEntityByTypeAndId( type, entityId ); if ( change.isRemove() ) { if ( null != entity ) { Disposable.dispose( entity ); response.incEntityRemoveCount(); } } else { final EntityChangeData data = change.getData(); if ( null == entity ) { final String name = Replicant.areNamesEnabled() ? entitySchema.getName() + "/" + entityId : null; entity = getReplicantContext().getEntityService().findOrCreateEntity( name, type, entityId ); final Object userObject = entitySchema.getCreator().createEntity( entityId, data ); entity.setUserObject( userObject ); } else { @SuppressWarnings( "rawtypes" ) final EntitySchema.Updater updater = entitySchema.getUpdater(); if ( null != updater ) { updater.updateEntity( entity.getUserObject(), data ); } } final String[] channels = change.getChannels(); final int schemaId = getSchema().getId(); for ( final String channel : channels ) { final int separator = channel.indexOf( "." ); final ChannelAddress address = new ChannelAddress( schemaId, Integer.parseInt( -1 == separator ? channel : channel.substring( 0, separator ) ), -1 == separator ? null : Integer.parseInt( channel.substring( separator + 1 ) ) ); final Subscription subscription = getReplicantContext().findSubscription( address ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> null != subscription, () -> "Replicant-0069: ChangeSet contained an EntityChange message referencing channel " + address + " but no such subscription exists locally." ); } assert null != subscription; entity.tryLinkToSubscription( subscription ); } response.incEntityUpdateCount(); response.changeProcessed( entity.getUserObject() ); } } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString(); }
Connector extends ReplicantService { @SuppressWarnings( "unchecked" ) @Action void processEntityChanges() { final MessageResponse response = ensureCurrentMessageResponse(); EntityChange change; for ( int i = 0; i < _changesToProcessPerTick && null != ( change = response.nextEntityChange() ); i++ ) { final String id = change.getId(); final int idSeparator = id.indexOf( "." ); final int typeId = Integer.parseInt( id.substring( 0, idSeparator ) ); final int entityId = Integer.parseInt( id.substring( idSeparator + 1 ) ); final EntitySchema entitySchema = getSchema().getEntity( typeId ); final Class<?> type = entitySchema.getType(); Entity entity = getReplicantContext().getEntityService().findEntityByTypeAndId( type, entityId ); if ( change.isRemove() ) { if ( null != entity ) { Disposable.dispose( entity ); response.incEntityRemoveCount(); } } else { final EntityChangeData data = change.getData(); if ( null == entity ) { final String name = Replicant.areNamesEnabled() ? entitySchema.getName() + "/" + entityId : null; entity = getReplicantContext().getEntityService().findOrCreateEntity( name, type, entityId ); final Object userObject = entitySchema.getCreator().createEntity( entityId, data ); entity.setUserObject( userObject ); } else { @SuppressWarnings( "rawtypes" ) final EntitySchema.Updater updater = entitySchema.getUpdater(); if ( null != updater ) { updater.updateEntity( entity.getUserObject(), data ); } } final String[] channels = change.getChannels(); final int schemaId = getSchema().getId(); for ( final String channel : channels ) { final int separator = channel.indexOf( "." ); final ChannelAddress address = new ChannelAddress( schemaId, Integer.parseInt( -1 == separator ? channel : channel.substring( 0, separator ) ), -1 == separator ? null : Integer.parseInt( channel.substring( separator + 1 ) ) ); final Subscription subscription = getReplicantContext().findSubscription( address ); if ( Replicant.shouldCheckInvariants() ) { invariant( () -> null != subscription, () -> "Replicant-0069: ChangeSet contained an EntityChange message referencing channel " + address + " but no such subscription exists locally." ); } assert null != subscription; entity.tryLinkToSubscription( subscription ); } response.incEntityUpdateCount(); response.changeProcessed( entity.getUserObject() ); } } } Connector( @Nullable final ReplicantContext context,
             @Nonnull final SystemSchema schema,
             @Nonnull final Transport transport ); @Override String toString();  }