import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
@Override
public Resource<Pod> getResource(ObserverKey observerKey, KubernetesClient client) {
    return new Resource<Pod>() {
        @Override
        public Watch watchResources(Watcher watcher) {
            return client.pods().inNamespace(client.getNamespace()).withLabels(observerKey.getLabelFilter()).watch(watcher);
        }

        @Override
        public Set<Pod> listResources() {
            return client.pods().inNamespace(client.getNamespace()).withLabels(observerKey.getLabelFilter()).list().getItems().stream()
                    .filter(pod -> filterPod(observerKey, pod))
                    .map(Pod::new)
                    .collect(Collectors.toSet());
        }
    };
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
@Override
public Resource<TestResource> getResource(ObserverKey observerKey, KubernetesClient client) {
    return new Resource<TestResource>() {
        @Override
        public Watch watchResources(Watcher watcher) {
            return testWatch;
        }

        @Override
        public Set<TestResource> listResources() {
            try {
                return stream.take();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    };
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
private CompletableFuture<Void> doDelete(String name) throws InfrastructureException {
  try {
    final PodResource<Pod, DoneablePod> podResource =
        clientFactory.create().pods().inNamespace(namespace).withName(name);
    final CompletableFuture<Void> deleteFuture = new CompletableFuture<>();
    final Watch watch = podResource.watch(new DeleteWatcher(deleteFuture));

    podResource.delete();
    return deleteFuture.whenComplete(
        (v, e) -> {
          if (e != null) {
            LOG.warn("Failed to remove pod {} cause {}", name, e.getMessage());
          }
          watch.close();
        });
  } catch (KubernetesClientException ex) {
    throw new InfrastructureException(ex.getMessage(), ex);
  }
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
/**
 * Returns the {@link PodList} that match the specified {@link Deployment}.
 *
 * @param deployment The {@link Deployment}
 */
public static PodList findMatching(KubernetesClient client, Deployment deployment) {
    FilterWatchListDeletable<Pod, PodList, Boolean, Watch, Watcher<Pod>> podLister =
            client.pods().inNamespace(deployment.getMetadata().getNamespace());
    if (deployment.getSpec().getSelector().getMatchLabels() != null) {
        podLister.withLabels(deployment.getSpec().getSelector().getMatchLabels());
    }
    if (deployment.getSpec().getSelector().getMatchExpressions() != null) {
        for (LabelSelectorRequirement req : deployment.getSpec().getSelector().getMatchExpressions()) {
            switch (req.getOperator()) {
                case "In":
                    podLister.withLabelIn(req.getKey(), req.getValues().toArray(new String[]{}));
                    break;
                case "NotIn":
                    podLister.withLabelNotIn(req.getKey(), req.getValues().toArray(new String[]{}));
                    break;
                case "DoesNotExist":
                    podLister.withoutLabel(req.getKey());
                    break;
                case "Exists":
                    podLister.withLabel(req.getKey());
                    break;
            }
        }
    }
    return podLister.list();
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
public synchronized void stop() {
    if (relister != null && !relister.isDone()) {
        relister.cancel(true);
        relister = null;
    }

    for (Map.Entry<String, Watch> entry : watches.entrySet()) {
        entry.getValue().close();
        watches.remove(entry.getKey());
    }
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
@PreDestroy
public void unwatch() {
    if (this.watches != null) {
        for (Map.Entry<String, Watch> entry : this.watches.entrySet()) {
            try {
                log.debug("Closing the watch {}", entry.getKey());
                entry.getValue().close();

            } catch (Exception e) {
                log.error("Error while closing the watch connection", e);
            }
        }
    }
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
private Watch registerWatcher() {
    return kubernetesClient.pods().watch(new Watcher<Pod>() {
        @Override
        public void eventReceived(Action eventAction, Pod eventPod) {
            boolean continueBlocking = continueBlocking(eventAction, eventPod);
            if (!continueBlocking)
                block[0] = false;
        }

        @Override
        public void onClose(KubernetesClientException e) {
        }
    });
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
private static ServiceList getServicesByLabel(Map<String, String> labels, KubernetesClient client)
    throws KubernetesClientException {
  Objects.requireNonNull(client, "no client available");
  final MixedOperation<Service, ServiceList, DoneableService, Resource<Service, DoneableService>>
      services = client.services();
  final FilterWatchListDeletable<Service, ServiceList, Boolean, Watch, Watcher<Service>>
      listable = createLabelFilterQuery(labels, services);
  return listable.list();
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
private static FilterWatchListDeletable<Service, ServiceList, Boolean, Watch, Watcher<Service>>
    createLabelFilterQuery(
        Map<String, String> labels,
        MixedOperation<Service, ServiceList, DoneableService, Resource<Service, DoneableService>>
            services) {
  FilterWatchListDeletable<Service, ServiceList, Boolean, Watch, Watcher<Service>> listable =
      null;
  for (Entry<String, String> entry : labels.entrySet()) {
    listable =
        listable == null
            ? services.withLabel(entry.getKey(), entry.getValue())
            : listable.withLabel(entry.getKey(), entry.getValue());
  }
  return listable;
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
@SuppressFBWarnings("EI_EXPOSE_REP2")
public BaseWatcher(String[] namespaces) {
    this.namespaces = namespaces;
    watches = new HashMap<String, Watch>();
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
public Watch watchPods(String namespace, Watcher<Pod> podWatcher) {
    return client.pods().inNamespace(namespace).watch(podWatcher);
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
@Override
public Watch watchResources(Watcher watcher) {
    return () -> {};
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
@Override
public Watch watchResources(io.fabric8.kubernetes.client.Watcher watcher) {
    return client.pods().withLabels(labelFilter).watch(watcher);
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
protected void addClient(Watch watch) {
    watches.add(watch);
}
 

import io.fabric8.kubernetes.client.Watch; //导入依赖的package包/类
private static List<String> print(FilterWatchListDeletable<Pod, PodList, Boolean, Watch, Watcher<Pod>> pods) {
    return pods.list().getItems().stream()
            .map(pod -> String.format("%s (%s)", pod.getMetadata().getName(), pod.getStatus().getPhase()))
            .collect(Collectors.toList());
}