@Test public void convergeAreaOfInterest() { final Connector connector = createConnector(); newConnection( connector ); pauseScheduler(); connector.pauseMessageScheduler(); safeAction( () -> connector.setState( ConnectorState.CONNECTED ) ); final ChannelAddress address = new ChannelAddress( 1, 0 ); final AreaOfInterest areaOfInterest = safeAction( () -> Replicant.context().createOrUpdateAreaOfInterest( address, null ) ); final TestSpyEventHandler handler = registerTestSpyEventHandler(); final Converger.Action result = Replicant.context().getConverger().convergeAreaOfInterest( areaOfInterest, null, null ); assertEquals( result, Converger.Action.SUBMITTED_ADD ); handler.assertEventCount( 1 ); handler.assertNextEvent( SubscribeRequestQueuedEvent.class, e -> assertEquals( e.getAddress(), address ) ); }
@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 );   }