import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
static<T> ObjectNode writeInstance(ObjectMapper mapper, ServiceInstance<T> instance, DiscoveryContext<T> context)
{
    ObjectNode  node = mapper.createObjectNode();
    node.put("name", instance.getName());
    node.put("id", instance.getId());
    node.put("address", instance.getAddress());
    putInteger(node, "port", instance.getPort());
    putInteger(node, "sslPort", instance.getSslPort());
    node.put("registrationTimeUTC", instance.getRegistrationTimeUTC());
    node.put("serviceType", instance.getServiceType().name());
    try
    {
        context.marshallJson(node, "payload", instance.getPayload());
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        throw new WebApplicationException(e);
    }

    return node;
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
@Override
public ServiceInstances<T> readFrom(Class<ServiceInstances<T>> type, Type genericType, Annotation[] annotations, MediaType mediaType, MultivaluedMap<String, String> httpHeaders, InputStream entityStream) throws IOException, WebApplicationException
{
    try
    {
        List<ServiceInstance<T>>    instances = Lists.newArrayList();
        ObjectMapper                mapper = new ObjectMapper();
        JsonNode                    tree = mapper.reader().readTree(entityStream);
        for ( int i = 0; i < tree.size(); ++i )
        {
            JsonNode                    node = tree.get(i);
            ServiceInstance<T> instance = JsonServiceInstanceMarshaller.readInstance(node, context);
            instances.add(instance);
        }
        return new ServiceInstances<T>(instances);
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        throw new WebApplicationException(e);
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
@DELETE
@Path("v1/service/{name}/{id}")
public Response     removeService(@PathParam("name") String name, @PathParam("id") String id)
{
    try
    {
        ServiceInstance<T> instance = context.getServiceDiscovery().queryForInstance(name, id);
        if ( instance != null )
        {
            //noinspection unchecked
            context.getServiceDiscovery().unregisterService(instance);
        }
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        log.error("Trying to delete service", e);
        return Response.serverError().build();
    }
    return Response.ok().build();
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
@GET
@Path("v1/service/{name}")
@Produces(MediaType.APPLICATION_JSON)
public Response     getAll(@PathParam("name") String name)
{
    try
    {
        Collection<ServiceInstance<T>>  instances = context.getServiceDiscovery().queryForInstances(name);
        return Response.ok(new ServiceInstances<T>(instances)).build();
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        log.error(String.format("Trying to get instances from service (%s)", name), e);
        return Response.serverError().build();
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
private Response internalGet(String name, String id, boolean addDeprecationHeader)
{
    try
    {
        ServiceInstance<T> instance = context.getServiceDiscovery().queryForInstance(name, id);
        if ( instance == null )
        {
            return Response.status(Response.Status.NOT_FOUND).build();
        }
        Response.ResponseBuilder builder = Response.ok(instance);
        if ( addDeprecationHeader )
        {
            builder = builder.header("Warning", "This API has been deprecated. Please see the updated spec for the replacement API.");
        }
        return builder.build();
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        log.error(String.format("Trying to get instance (%s) from service (%s)", id, name), e);
        return Response.serverError().build();
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
private void checkService(String name)
{
    try
    {
        Collection<ServiceInstance<Object>>     instances = discovery.queryForInstances(name);
        for ( ServiceInstance<Object> instance : instances )
        {
            if ( instance.getServiceType() == ServiceType.STATIC )
            {
                if ( (System.currentTimeMillis() - instance.getRegistrationTimeUTC()) > instanceRefreshMs )
                {
                    discovery.unregisterService(instance);
                }
            }
        }
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        log.error(String.format("GC for service: %s", name), e);
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
void addFailedOperation(T details)
{
    if ( debugListener != null )
    {
        debugListener.pathAddedForGuaranteedOperation(details);
    }
    
    
    if ( client.getState() == CuratorFrameworkState.STARTED )
    {
        log.debug("Details being added to guaranteed operation set: " + details);
        try
        {
            executeGuaranteedOperationInBackground(details);
        }
        catch ( Exception e )
        {
            ThreadUtils.checkInterrupted(e);
            addFailedOperation(details);
        }
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
/**
 * Utility - apply the given function to each listener. The function receives
 * the listener as an argument.
 *
 * @param function function to call for each listener
 */
public void     forEach(final Function<T, Void> function)
{
    for ( final ListenerEntry<T> entry : listeners.values() )
    {
        entry.executor.execute
        (
            new Runnable()
            {
                @Override
                public void run()
                {
                    try
                    {
                        function.apply(entry.listener);
                    }
                    catch ( Throwable e )
                    {
                        ThreadUtils.checkInterrupted(e);
                        log.error(String.format("Listener (%s) threw an exception", entry.listener), e);
                    }
                }
            }
        );
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
@Override
public void childEvent(CuratorFramework client, TreeCacheEvent event)
{
    try
    {
        internalChildEvent(event);
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);

        listenerContainer.forEach(l -> {
            l.handleException(e);
            return null;
        });
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
private void notifyListeners()
{
    final byte[] localValue = getValue();
    listeners.forEach
        (
            new Function<SharedValueListener, Void>()
            {
                @Override
                public Void apply(SharedValueListener listener)
                {
                    try
                    {
                        listener.valueHasChanged(SharedValue.this, localValue);
                    }
                    catch ( Exception e )
                    {
                        ThreadUtils.checkInterrupted(e);
                        log.error("From SharedValue listener", e);
                    }
                    return null;
                }
            }
        );
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
private void notifyListeners()
{
    final String path = getActualPath();
    listeners.forEach(
         new Function<PersistentNodeListener, Void>()
         {
             @Override
             public Void apply(PersistentNodeListener listener)
             {
                 try
                {
                    listener.nodeCreated(path);
                }
                catch ( Exception e )
                {
                    ThreadUtils.checkInterrupted(e);
                    log.error("From PersistentNode listener", e);
                }
                return null;
            }
         }
    );
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
void callListeners(final PathChildrenCacheEvent event)
{
    listeners.forEach
        (
            new Function<PathChildrenCacheListener, Void>()
            {
                @Override
                public Void apply(PathChildrenCacheListener listener)
                {
                    try
                    {
                        listener.childEvent(client, event);
                    }
                    catch ( Exception e )
                    {
                        ThreadUtils.checkInterrupted(e);
                        handleException(e);
                    }
                    return null;
                }
            }
        );
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
/**
 * Close/end the cache.
 */
@Override
public void close()
{
    if ( treeState.compareAndSet(TreeState.STARTED, TreeState.CLOSED) )
    {
        client.removeWatchers();
        client.getConnectionStateListenable().removeListener(connectionStateListener);
        listeners.clear();
        executorService.shutdown();
        try
        {
            root.wasDeleted();
        }
        catch ( Exception e )
        {
            ThreadUtils.checkInterrupted(e);
            handleException(e);
        }
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
private void callListeners(final TreeCacheEvent event)
{
    listeners.forEach(new Function<TreeCacheListener, Void>()
    {
        @Override
        public Void apply(TreeCacheListener listener)
        {
            try
            {
                listener.childEvent(client, event);
            }
            catch ( Exception e )
            {
                ThreadUtils.checkInterrupted(e);
                handleException(e);
            }
            return null;
        }
    });
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
private void publishEvent(final TreeCacheEvent event)
{
    if ( treeState.get() != TreeState.CLOSED )
    {
        LOG.debug("publishEvent: {}", event);
        executorService.submit(new Runnable()
        {
            @Override
            public void run()
            {
                {
                    try
                    {
                        callListeners(event);
                    }
                    catch ( Exception e )
                    {
                        ThreadUtils.checkInterrupted(e);
                        handleException(e);
                    }
                }
            }
        });
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
private synchronized void internalStart()
{
    if ( state.get() == State.STARTED )
    {
        client.getConnectionStateListenable().addListener(listener);
        try
        {
            reset();
        }
        catch ( Exception e )
        {
            ThreadUtils.checkInterrupted(e);
            log.error("An error occurred checking resetting leadership.", e);
        }
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
/**
 * <p>Acquire <code>qty</code> leases. If there are not enough leases available, this method
 * blocks until either the maximum number of leases is increased enough or other clients/processes
 * close enough leases.</p>
 *
 * <p>The client must close the leases when it is done with them. You should do this in a
 * <code>finally</code> block. NOTE: You can use {@link #returnAll(Collection)} for this.</p>
 *
 * @param qty number of leases to acquire
 * @return the new leases
 * @throws Exception ZK errors, interruptions, etc.
 */
public Collection<Lease> acquire(int qty) throws Exception
{
    Preconditions.checkArgument(qty > 0, "qty cannot be 0");

    ImmutableList.Builder<Lease>    builder = ImmutableList.builder();
    try
    {
        while ( qty-- > 0 )
        {
            String      path = internals.attemptLock(-1, null, null);
            builder.add(makeLease(path));
        }
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        returnAll(builder.build());
        throw e;
    }
    return builder.build();
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
/**
 * Close the client
 */
public void close()
{
    log.debug("Closing");

    started.set(false);
    try
    {
        state.close();
    }
    catch ( IOException e )
    {
        ThreadUtils.checkInterrupted(e);
        log.error("", e);
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
@Override
public void close() throws IOException
{
    log.debug("Closing");

    CloseableUtils.closeQuietly(ensembleProvider);
    try
    {
        zooKeeper.closeAndClear();
    }
    catch ( Exception e )
    {
        ThreadUtils.checkInterrupted(e);
        throw new IOException(e);
    }
    finally
    {
        isConnected.set(false);
    }
}
 

import org.apache.curator.utils.ThreadUtils; //导入依赖的package包/类
@Override
public <T> T callWithRetry(CuratorZookeeperClient client, Callable<T> proc) throws Exception
{
    client.internalBlockUntilConnectedOrTimedOut();

    T result = null;
    RetryLoop retryLoop = client.newRetryLoop();
    while ( retryLoop.shouldContinue() )
    {
        try
        {
            result = proc.call();
            retryLoop.markComplete();
        }
        catch ( Exception e )
        {
            ThreadUtils.checkInterrupted(e);
            retryLoop.takeException(e);
        }
    }

    return result;
}