@Test public void convergeAreaOfInterest_typeDiffers() { final ChannelSchema channel0 = new ChannelSchema( 0, ValueUtil.randomString(), null, ChannelSchema.FilterType.DYNAMIC, ( f, e ) -> true, true, true, Collections.emptyList() ); final SystemSchema schema = new SystemSchema( 1, ValueUtil.randomString(), new ChannelSchema[]{ channel0 }, new EntitySchema[ 0 ] ); final Connector connector = createConnector( schema ); newConnection( connector ); safeAction( () -> connector.setState( ConnectorState.CONNECTED ) ); final ChannelAddress address1 = new ChannelAddress( 1, 0, 1 ); final ChannelAddress address2 = new ChannelAddress( 1, 0, 2 ); final AreaOfInterest areaOfInterest1 = safeAction( () -> Replicant.context().createOrUpdateAreaOfInterest( address1, null ) ); final AreaOfInterest areaOfInterest2 = safeAction( () -> Replicant.context().createOrUpdateAreaOfInterest( address2, null ) ); createSubscription( address2, ValueUtil.randomString(), true ); final TestSpyEventHandler handler = registerTestSpyEventHandler(); final Converger.Action result = Replicant.context() .getConverger() .convergeAreaOfInterest( areaOfInterest2, areaOfInterest1, AreaOfInterestRequest.Type.ADD ); assertEquals( result, Converger.Action.NO_ACTION ); handler.assertEventCount( 0 ); }
@arez.annotations.Action( requireNewTransaction = true, verifyRequired = false ) @Nonnull Action convergeAreaOfInterest( @Nonnull final AreaOfInterest areaOfInterest, @Nullable final AreaOfInterest groupTemplate, @Nullable final AreaOfInterestRequest.Type groupAction ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> Disposable.isNotDisposed( areaOfInterest ), () -> "Replicant-0020: Invoked convergeAreaOfInterest() with disposed AreaOfInterest." ); } final ChannelAddress address = areaOfInterest.getAddress(); final Connector connector = getReplicantRuntime().getConnector( address.getSystemId() ); if ( ConnectorState.CONNECTED == connector.getState() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); final boolean subscribed = null != subscription; final Object filter = areaOfInterest.getFilter(); final int addIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.ADD, address, filter ); final int removeIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.REMOVE, address, null ); final int updateIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.UPDATE, address, filter ); if ( ( !subscribed && addIndex < 0 ) || removeIndex > addIndex ) { if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.ADD ) ) { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } else { return Action.NO_ACTION; } } else if ( addIndex >= 0 ) { return Action.IN_PROGRESS; } else { if ( updateIndex >= 0 ) { return Action.IN_PROGRESS; } if ( !FilterUtil.filtersEqual( filter, subscription.getFilter() ) ) { final SystemSchema schema = getReplicantContext().getSchemaService().getById( address.getSystemId() ); final ChannelSchema.FilterType filterType = schema.getChannel( address.getChannelId() ).getFilterType(); if ( null == groupTemplate && ChannelSchema.FilterType.DYNAMIC != filterType ) { if ( Replicant.shouldCheckInvariants() ) { invariant( subscription::isExplicitSubscription, () -> "Replicant-0083: Attempting to update channel " + address + " but channel does not " + "allow dynamic updates of filter and channel has not been explicitly subscribed." ); } connector.requestUnsubscribe( address ); return Action.SUBMITTED_REMOVE; } else if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.UPDATE ) ) { connector.requestSubscriptionUpdate( address, filter ); return Action.SUBMITTED_UPDATE; } else { return Action.NO_ACTION; } } else { if ( AreaOfInterest.Status.NOT_ASKED == areaOfInterest.getStatus() ) { if ( subscription.isExplicitSubscription() ) { areaOfInterest.updateAreaOfInterest( AreaOfInterest.Status.LOADED, null ); } else { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } } } } } return Action.NO_ACTION; }
Converger extends ReplicantService { @arez.annotations.Action( requireNewTransaction = true, verifyRequired = false ) @Nonnull Action convergeAreaOfInterest( @Nonnull final AreaOfInterest areaOfInterest, @Nullable final AreaOfInterest groupTemplate, @Nullable final AreaOfInterestRequest.Type groupAction ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> Disposable.isNotDisposed( areaOfInterest ), () -> "Replicant-0020: Invoked convergeAreaOfInterest() with disposed AreaOfInterest." ); } final ChannelAddress address = areaOfInterest.getAddress(); final Connector connector = getReplicantRuntime().getConnector( address.getSystemId() ); if ( ConnectorState.CONNECTED == connector.getState() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); final boolean subscribed = null != subscription; final Object filter = areaOfInterest.getFilter(); final int addIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.ADD, address, filter ); final int removeIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.REMOVE, address, null ); final int updateIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.UPDATE, address, filter ); if ( ( !subscribed && addIndex < 0 ) || removeIndex > addIndex ) { if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.ADD ) ) { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } else { return Action.NO_ACTION; } } else if ( addIndex >= 0 ) { return Action.IN_PROGRESS; } else { if ( updateIndex >= 0 ) { return Action.IN_PROGRESS; } if ( !FilterUtil.filtersEqual( filter, subscription.getFilter() ) ) { final SystemSchema schema = getReplicantContext().getSchemaService().getById( address.getSystemId() ); final ChannelSchema.FilterType filterType = schema.getChannel( address.getChannelId() ).getFilterType(); if ( null == groupTemplate && ChannelSchema.FilterType.DYNAMIC != filterType ) { if ( Replicant.shouldCheckInvariants() ) { invariant( subscription::isExplicitSubscription, () -> "Replicant-0083: Attempting to update channel " + address + " but channel does not " + "allow dynamic updates of filter and channel has not been explicitly subscribed." ); } connector.requestUnsubscribe( address ); return Action.SUBMITTED_REMOVE; } else if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.UPDATE ) ) { connector.requestSubscriptionUpdate( address, filter ); return Action.SUBMITTED_UPDATE; } else { return Action.NO_ACTION; } } else { if ( AreaOfInterest.Status.NOT_ASKED == areaOfInterest.getStatus() ) { if ( subscription.isExplicitSubscription() ) { areaOfInterest.updateAreaOfInterest( AreaOfInterest.Status.LOADED, null ); } else { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } } } } } return Action.NO_ACTION; } }
Converger extends ReplicantService { @arez.annotations.Action( requireNewTransaction = true, verifyRequired = false ) @Nonnull Action convergeAreaOfInterest( @Nonnull final AreaOfInterest areaOfInterest, @Nullable final AreaOfInterest groupTemplate, @Nullable final AreaOfInterestRequest.Type groupAction ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> Disposable.isNotDisposed( areaOfInterest ), () -> "Replicant-0020: Invoked convergeAreaOfInterest() with disposed AreaOfInterest." ); } final ChannelAddress address = areaOfInterest.getAddress(); final Connector connector = getReplicantRuntime().getConnector( address.getSystemId() ); if ( ConnectorState.CONNECTED == connector.getState() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); final boolean subscribed = null != subscription; final Object filter = areaOfInterest.getFilter(); final int addIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.ADD, address, filter ); final int removeIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.REMOVE, address, null ); final int updateIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.UPDATE, address, filter ); if ( ( !subscribed && addIndex < 0 ) || removeIndex > addIndex ) { if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.ADD ) ) { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } else { return Action.NO_ACTION; } } else if ( addIndex >= 0 ) { return Action.IN_PROGRESS; } else { if ( updateIndex >= 0 ) { return Action.IN_PROGRESS; } if ( !FilterUtil.filtersEqual( filter, subscription.getFilter() ) ) { final SystemSchema schema = getReplicantContext().getSchemaService().getById( address.getSystemId() ); final ChannelSchema.FilterType filterType = schema.getChannel( address.getChannelId() ).getFilterType(); if ( null == groupTemplate && ChannelSchema.FilterType.DYNAMIC != filterType ) { if ( Replicant.shouldCheckInvariants() ) { invariant( subscription::isExplicitSubscription, () -> "Replicant-0083: Attempting to update channel " + address + " but channel does not " + "allow dynamic updates of filter and channel has not been explicitly subscribed." ); } connector.requestUnsubscribe( address ); return Action.SUBMITTED_REMOVE; } else if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.UPDATE ) ) { connector.requestSubscriptionUpdate( address, filter ); return Action.SUBMITTED_UPDATE; } else { return Action.NO_ACTION; } } else { if ( AreaOfInterest.Status.NOT_ASKED == areaOfInterest.getStatus() ) { if ( subscription.isExplicitSubscription() ) { areaOfInterest.updateAreaOfInterest( AreaOfInterest.Status.LOADED, null ); } else { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } } } } } return Action.NO_ACTION; } Converger( @Nullable final ReplicantContext context ); }
Converger extends ReplicantService { @arez.annotations.Action( requireNewTransaction = true, verifyRequired = false ) @Nonnull Action convergeAreaOfInterest( @Nonnull final AreaOfInterest areaOfInterest, @Nullable final AreaOfInterest groupTemplate, @Nullable final AreaOfInterestRequest.Type groupAction ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> Disposable.isNotDisposed( areaOfInterest ), () -> "Replicant-0020: Invoked convergeAreaOfInterest() with disposed AreaOfInterest." ); } final ChannelAddress address = areaOfInterest.getAddress(); final Connector connector = getReplicantRuntime().getConnector( address.getSystemId() ); if ( ConnectorState.CONNECTED == connector.getState() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); final boolean subscribed = null != subscription; final Object filter = areaOfInterest.getFilter(); final int addIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.ADD, address, filter ); final int removeIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.REMOVE, address, null ); final int updateIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.UPDATE, address, filter ); if ( ( !subscribed && addIndex < 0 ) || removeIndex > addIndex ) { if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.ADD ) ) { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } else { return Action.NO_ACTION; } } else if ( addIndex >= 0 ) { return Action.IN_PROGRESS; } else { if ( updateIndex >= 0 ) { return Action.IN_PROGRESS; } if ( !FilterUtil.filtersEqual( filter, subscription.getFilter() ) ) { final SystemSchema schema = getReplicantContext().getSchemaService().getById( address.getSystemId() ); final ChannelSchema.FilterType filterType = schema.getChannel( address.getChannelId() ).getFilterType(); if ( null == groupTemplate && ChannelSchema.FilterType.DYNAMIC != filterType ) { if ( Replicant.shouldCheckInvariants() ) { invariant( subscription::isExplicitSubscription, () -> "Replicant-0083: Attempting to update channel " + address + " but channel does not " + "allow dynamic updates of filter and channel has not been explicitly subscribed." ); } connector.requestUnsubscribe( address ); return Action.SUBMITTED_REMOVE; } else if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.UPDATE ) ) { connector.requestSubscriptionUpdate( address, filter ); return Action.SUBMITTED_UPDATE; } else { return Action.NO_ACTION; } } else { if ( AreaOfInterest.Status.NOT_ASKED == areaOfInterest.getStatus() ) { if ( subscription.isExplicitSubscription() ) { areaOfInterest.updateAreaOfInterest( AreaOfInterest.Status.LOADED, null ); } else { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } } } } } return Action.NO_ACTION; } Converger( @Nullable final ReplicantContext context );  }
Converger extends ReplicantService { @arez.annotations.Action( requireNewTransaction = true, verifyRequired = false ) @Nonnull Action convergeAreaOfInterest( @Nonnull final AreaOfInterest areaOfInterest, @Nullable final AreaOfInterest groupTemplate, @Nullable final AreaOfInterestRequest.Type groupAction ) { if ( Replicant.shouldCheckInvariants() ) { invariant( () -> Disposable.isNotDisposed( areaOfInterest ), () -> "Replicant-0020: Invoked convergeAreaOfInterest() with disposed AreaOfInterest." ); } final ChannelAddress address = areaOfInterest.getAddress(); final Connector connector = getReplicantRuntime().getConnector( address.getSystemId() ); if ( ConnectorState.CONNECTED == connector.getState() ) { final Subscription subscription = getReplicantContext().findSubscription( address ); final boolean subscribed = null != subscription; final Object filter = areaOfInterest.getFilter(); final int addIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.ADD, address, filter ); final int removeIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.REMOVE, address, null ); final int updateIndex = connector.lastIndexOfPendingAreaOfInterestRequest( AreaOfInterestRequest.Type.UPDATE, address, filter ); if ( ( !subscribed && addIndex < 0 ) || removeIndex > addIndex ) { if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.ADD ) ) { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } else { return Action.NO_ACTION; } } else if ( addIndex >= 0 ) { return Action.IN_PROGRESS; } else { if ( updateIndex >= 0 ) { return Action.IN_PROGRESS; } if ( !FilterUtil.filtersEqual( filter, subscription.getFilter() ) ) { final SystemSchema schema = getReplicantContext().getSchemaService().getById( address.getSystemId() ); final ChannelSchema.FilterType filterType = schema.getChannel( address.getChannelId() ).getFilterType(); if ( null == groupTemplate && ChannelSchema.FilterType.DYNAMIC != filterType ) { if ( Replicant.shouldCheckInvariants() ) { invariant( subscription::isExplicitSubscription, () -> "Replicant-0083: Attempting to update channel " + address + " but channel does not " + "allow dynamic updates of filter and channel has not been explicitly subscribed." ); } connector.requestUnsubscribe( address ); return Action.SUBMITTED_REMOVE; } else if ( null == groupTemplate || canGroup( groupTemplate, groupAction, areaOfInterest, AreaOfInterestRequest.Type.UPDATE ) ) { connector.requestSubscriptionUpdate( address, filter ); return Action.SUBMITTED_UPDATE; } else { return Action.NO_ACTION; } } else { if ( AreaOfInterest.Status.NOT_ASKED == areaOfInterest.getStatus() ) { if ( subscription.isExplicitSubscription() ) { areaOfInterest.updateAreaOfInterest( AreaOfInterest.Status.LOADED, null ); } else { connector.requestSubscribe( address, filter ); return Action.SUBMITTED_ADD; } } } } } return Action.NO_ACTION; } Converger( @Nullable final ReplicantContext context );   }