DefaultAgentCache implements AgentCache { @Override public Single<AgentInstance> updateAgentInstanceStore(String instanceId, Function<AgentInstance, AgentInstance> function) { Single<AgentInstance> single = onEventLoopWithSubscription(() -> { AgentInstance agentInstance = getAgentInstance(instanceId); AgentInstance updatedAgentInstance = function.apply(agentInstance); setDataSnapshot(dataSnapshot.updateAgentInstance(updatedAgentInstance)); eventSubject.onNext(new CacheUpdateEvent(CacheUpdateType.Instance, instanceId)); return updatedAgentInstance; }); return single.flatMap(ai -> agentStore.storeAgentInstance(ai).toSingle(() -> ai)); } @Inject  DefaultAgentCache(AgentManagementConfiguration configuration,
                             AgentStore agentStore,
                             InstanceCloudConnector connector,
                             Registry registry);   DefaultAgentCache(AgentManagementConfiguration configuration,
                      AgentStore agentStore,
                      InstanceCloudConnector connector,
                      Registry registry,
                      Scheduler scheduler); @Activator void enterActiveMode(); void shutdown(); @Override List<AgentInstanceGroup> getInstanceGroups(); @Override AgentInstanceGroup getInstanceGroup(String instanceGroupId); @Override Optional<AgentInstanceGroup> findInstanceGroup(String instanceGroupId); @Override Set<AgentInstance> getAgentInstances(String instanceGroupId); @Override AgentInstance getAgentInstance(String instanceId); @Override Optional<AgentInstance> findAgentInstance(String instanceId); @Override Single<AgentInstanceGroup> updateInstanceGroupStore(String instanceGroupId, Function<AgentInstanceGroup, AgentInstanceGroup> function); @Override Single<AgentInstanceGroup> updateInstanceGroupStoreAndSyncCloud(String instanceGroupId, Function<AgentInstanceGroup, AgentInstanceGroup> function); @Override Single<AgentInstance> updateAgentInstanceStore(String instanceId, Function<AgentInstance, AgentInstance> function); @Override Completable removeInstances(String instanceGroupId, Set<String> agentInstanceIds); @Override Observable<CacheUpdateEvent> events();  }
@Test public void testUpdateAgentInstanceStore() { AgentInstanceGroup instanceGroup = cache.getInstanceGroups().get(0); String instanceGroupId = instanceGroup.getId(); AgentInstance agentInstance = CollectionsExt.first(cache.getAgentInstances(instanceGroupId)); assertThat(agentInstance).isNotNull(); String instanceId = agentInstance.getId(); ExtTestSubscriber<Object> testSubscriber = new ExtTestSubscriber<>(); Map<String, String> updatedAttributes = CollectionsExt.merge(agentInstance.getAttributes(), Collections.singletonMap("a", "1")); cache.updateAgentInstanceStore(instanceId, ig -> ig.toBuilder().withAttributes(updatedAttributes).build() ).toObservable().subscribe(testSubscriber); testScheduler.triggerActions(); AgentInstance cachedAgentInstance = cache.getAgentInstance(instanceId); assertThat(cachedAgentInstance.getAttributes()).containsKey("a").containsValue("1"); }