@Test public void testLookupMicroservice() throws Exception { Set<ServiceHandle> mockHandles = Util.createSetFrom(Util.createHandle(1), Util.createHandle(2)); Set<Object> services = Util.createSetFrom(new Object(), new Object()); new Expectations() {{ sender.sendToClusterAsync(META_DATA, (Set<Address>) any, (FutureListener<RspList<MicroserviceSearchResponse>>) any); times = 1; result = mockHandles; store.getServices(META_DATA); result = services; times = 1; }}; Set<Object> objects = clusterMicroserviceProvider.lookupMicroservice(META_DATA); assertThat(objects).isNotEmpty().isEqualTo(services); }
@Override public Set<Object> lookupMicroservice(final MicroserviceMetaData metaData) { try { Set<Address> addressesForMetadata = alreadyQueriedAddresses.getOrDefault(metaData, new HashSet<>()); this.sender.sendToClusterAsync(metaData, addressesForMetadata, new FutureListenerHelper<MicroserviceSearchResponse>(rspList -> { try { RspList<MicroserviceSearchResponse> responseRspList = rspList.get(10, TimeUnit.SECONDS); log.info("Response retrieved! {}", responseRspList); Collection<Rsp<MicroserviceSearchResponse>> result = responseRspList.values(); if (log.isTraceEnabled()) { log.trace("Size of a responses is : {} ", responseRspList.getResults().size()); } result.stream().filter(Rsp::hasException).forEach(rsp -> log.error("Exception was thrown during lookup on node: " + rsp.getSender(), rsp.getException())); Set<ServiceHandle> remoteServiceHandles = result.stream() .filter(rsp -> rsp.wasReceived() && !rsp.hasException() && rsp.getValue().getResult().canBeUsed()) .map((rsp) -> new RemoteServiceHandle(rsp.getSender(), rsp.getValue().getHandle(), sender, metaData)) .collect(Collectors.toSet()); addressesForMetadata.addAll(responseRspList.values().stream().map(Rsp::getSender).collect(Collectors.toSet())); alreadyQueriedAddresses.put(metaData, addressesForMetadata); this.remoteServiceHandlesStore.addHandles(metaData, remoteServiceHandles); } catch (Throwable e) { log.error("Error while looking up microservices.", e); } })); if (addressesForMetadata.isEmpty() && !sender.isEmptyCluster()) { Thread.sleep(timeout); } return this.remoteServiceHandlesStore.getServices(metaData); } catch (Throwable e) { log.error("Error while looking up microservices.", e); return emptySet(); } }
ClusterMicroserviceProvider implements MicroserviceProvider, ClusterSilverService { @Override public Set<Object> lookupMicroservice(final MicroserviceMetaData metaData) { try { Set<Address> addressesForMetadata = alreadyQueriedAddresses.getOrDefault(metaData, new HashSet<>()); this.sender.sendToClusterAsync(metaData, addressesForMetadata, new FutureListenerHelper<MicroserviceSearchResponse>(rspList -> { try { RspList<MicroserviceSearchResponse> responseRspList = rspList.get(10, TimeUnit.SECONDS); log.info("Response retrieved! {}", responseRspList); Collection<Rsp<MicroserviceSearchResponse>> result = responseRspList.values(); if (log.isTraceEnabled()) { log.trace("Size of a responses is : {} ", responseRspList.getResults().size()); } result.stream().filter(Rsp::hasException).forEach(rsp -> log.error("Exception was thrown during lookup on node: " + rsp.getSender(), rsp.getException())); Set<ServiceHandle> remoteServiceHandles = result.stream() .filter(rsp -> rsp.wasReceived() && !rsp.hasException() && rsp.getValue().getResult().canBeUsed()) .map((rsp) -> new RemoteServiceHandle(rsp.getSender(), rsp.getValue().getHandle(), sender, metaData)) .collect(Collectors.toSet()); addressesForMetadata.addAll(responseRspList.values().stream().map(Rsp::getSender).collect(Collectors.toSet())); alreadyQueriedAddresses.put(metaData, addressesForMetadata); this.remoteServiceHandlesStore.addHandles(metaData, remoteServiceHandles); } catch (Throwable e) { log.error("Error while looking up microservices.", e); } })); if (addressesForMetadata.isEmpty() && !sender.isEmptyCluster()) { Thread.sleep(timeout); } return this.remoteServiceHandlesStore.getServices(metaData); } catch (Throwable e) { log.error("Error while looking up microservices.", e); return emptySet(); } } }
ClusterMicroserviceProvider implements MicroserviceProvider, ClusterSilverService { @Override public Set<Object> lookupMicroservice(final MicroserviceMetaData metaData) { try { Set<Address> addressesForMetadata = alreadyQueriedAddresses.getOrDefault(metaData, new HashSet<>()); this.sender.sendToClusterAsync(metaData, addressesForMetadata, new FutureListenerHelper<MicroserviceSearchResponse>(rspList -> { try { RspList<MicroserviceSearchResponse> responseRspList = rspList.get(10, TimeUnit.SECONDS); log.info("Response retrieved! {}", responseRspList); Collection<Rsp<MicroserviceSearchResponse>> result = responseRspList.values(); if (log.isTraceEnabled()) { log.trace("Size of a responses is : {} ", responseRspList.getResults().size()); } result.stream().filter(Rsp::hasException).forEach(rsp -> log.error("Exception was thrown during lookup on node: " + rsp.getSender(), rsp.getException())); Set<ServiceHandle> remoteServiceHandles = result.stream() .filter(rsp -> rsp.wasReceived() && !rsp.hasException() && rsp.getValue().getResult().canBeUsed()) .map((rsp) -> new RemoteServiceHandle(rsp.getSender(), rsp.getValue().getHandle(), sender, metaData)) .collect(Collectors.toSet()); addressesForMetadata.addAll(responseRspList.values().stream().map(Rsp::getSender).collect(Collectors.toSet())); alreadyQueriedAddresses.put(metaData, addressesForMetadata); this.remoteServiceHandlesStore.addHandles(metaData, remoteServiceHandles); } catch (Throwable e) { log.error("Error while looking up microservices.", e); } })); if (addressesForMetadata.isEmpty() && !sender.isEmptyCluster()) { Thread.sleep(timeout); } return this.remoteServiceHandlesStore.getServices(metaData); } catch (Throwable e) { log.error("Error while looking up microservices.", e); return emptySet(); } }  }
ClusterMicroserviceProvider implements MicroserviceProvider, ClusterSilverService { @Override public Set<Object> lookupMicroservice(final MicroserviceMetaData metaData) { try { Set<Address> addressesForMetadata = alreadyQueriedAddresses.getOrDefault(metaData, new HashSet<>()); this.sender.sendToClusterAsync(metaData, addressesForMetadata, new FutureListenerHelper<MicroserviceSearchResponse>(rspList -> { try { RspList<MicroserviceSearchResponse> responseRspList = rspList.get(10, TimeUnit.SECONDS); log.info("Response retrieved! {}", responseRspList); Collection<Rsp<MicroserviceSearchResponse>> result = responseRspList.values(); if (log.isTraceEnabled()) { log.trace("Size of a responses is : {} ", responseRspList.getResults().size()); } result.stream().filter(Rsp::hasException).forEach(rsp -> log.error("Exception was thrown during lookup on node: " + rsp.getSender(), rsp.getException())); Set<ServiceHandle> remoteServiceHandles = result.stream() .filter(rsp -> rsp.wasReceived() && !rsp.hasException() && rsp.getValue().getResult().canBeUsed()) .map((rsp) -> new RemoteServiceHandle(rsp.getSender(), rsp.getValue().getHandle(), sender, metaData)) .collect(Collectors.toSet()); addressesForMetadata.addAll(responseRspList.values().stream().map(Rsp::getSender).collect(Collectors.toSet())); alreadyQueriedAddresses.put(metaData, addressesForMetadata); this.remoteServiceHandlesStore.addHandles(metaData, remoteServiceHandles); } catch (Throwable e) { log.error("Error while looking up microservices.", e); } })); if (addressesForMetadata.isEmpty() && !sender.isEmptyCluster()) { Thread.sleep(timeout); } return this.remoteServiceHandlesStore.getServices(metaData); } catch (Throwable e) { log.error("Error while looking up microservices.", e); return emptySet(); } }  @Override void initialize(final Context context); @Override Context getContext(); @Override void run(); @Override Set<Object> lookupMicroservice(final MicroserviceMetaData metaData); @Override Set<Object> lookupLocalMicroservice(final MicroserviceMetaData metaData); }
ClusterMicroserviceProvider implements MicroserviceProvider, ClusterSilverService { @Override public Set<Object> lookupMicroservice(final MicroserviceMetaData metaData) { try { Set<Address> addressesForMetadata = alreadyQueriedAddresses.getOrDefault(metaData, new HashSet<>()); this.sender.sendToClusterAsync(metaData, addressesForMetadata, new FutureListenerHelper<MicroserviceSearchResponse>(rspList -> { try { RspList<MicroserviceSearchResponse> responseRspList = rspList.get(10, TimeUnit.SECONDS); log.info("Response retrieved! {}", responseRspList); Collection<Rsp<MicroserviceSearchResponse>> result = responseRspList.values(); if (log.isTraceEnabled()) { log.trace("Size of a responses is : {} ", responseRspList.getResults().size()); } result.stream().filter(Rsp::hasException).forEach(rsp -> log.error("Exception was thrown during lookup on node: " + rsp.getSender(), rsp.getException())); Set<ServiceHandle> remoteServiceHandles = result.stream() .filter(rsp -> rsp.wasReceived() && !rsp.hasException() && rsp.getValue().getResult().canBeUsed()) .map((rsp) -> new RemoteServiceHandle(rsp.getSender(), rsp.getValue().getHandle(), sender, metaData)) .collect(Collectors.toSet()); addressesForMetadata.addAll(responseRspList.values().stream().map(Rsp::getSender).collect(Collectors.toSet())); alreadyQueriedAddresses.put(metaData, addressesForMetadata); this.remoteServiceHandlesStore.addHandles(metaData, remoteServiceHandles); } catch (Throwable e) { log.error("Error while looking up microservices.", e); } })); if (addressesForMetadata.isEmpty() && !sender.isEmptyCluster()) { Thread.sleep(timeout); } return this.remoteServiceHandlesStore.getServices(metaData); } catch (Throwable e) { log.error("Error while looking up microservices.", e); return emptySet(); } }  @Override void initialize(final Context context); @Override Context getContext(); @Override void run(); @Override Set<Object> lookupMicroservice(final MicroserviceMetaData metaData); @Override Set<Object> lookupLocalMicroservice(final MicroserviceMetaData metaData);  }