You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@curator.apache.org by ra...@apache.org on 2013/07/23 01:26:59 UTC
[24/32] git commit: wip
wip
Project: http://git-wip-us.apache.org/repos/asf/incubator-curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-curator/commit/46d112f8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-curator/tree/46d112f8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-curator/diff/46d112f8
Branch: refs/heads/CURATOR-14
Commit: 46d112f8fa7c39d8107220e5d74f17c92411ee0b
Parents: acf9c89
Author: randgalt <ra...@apache.org>
Authored: Tue Jun 25 08:39:14 2013 -0500
Committer: randgalt <ra...@apache.org>
Committed: Tue Jun 25 08:39:14 2013 -0500
----------------------------------------------------------------------
.../curator/x/discovery/DownInstancePolicy.java | 52 ++++++++++++++++++++
.../curator/x/discovery/ServiceProvider.java | 5 +-
.../x/discovery/ServiceProviderBuilder.java | 2 +-
.../discovery/details/DownInstanceManager.java | 28 +++--------
.../details/ServiceProviderBuilderImpl.java | 11 ++---
.../discovery/details/ServiceProviderImpl.java | 19 ++++++-
.../details/TestDownInstanceManager.java | 6 +--
7 files changed, 89 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-curator/blob/46d112f8/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/DownInstancePolicy.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/DownInstancePolicy.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/DownInstancePolicy.java
new file mode 100644
index 0000000..0360200
--- /dev/null
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/DownInstancePolicy.java
@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.curator.x.discovery;
+
+import java.util.concurrent.TimeUnit;
+
+public class DownInstancePolicy
+{
+ private final long timeoutMs;
+ private final int threshold;
+
+ private static final long DEFAULT_TIMEOUT_MS = 30000;
+ private static final int DEFAULT_THRESHOLD = 2;
+
+ public DownInstancePolicy()
+ {
+ this(DEFAULT_TIMEOUT_MS, TimeUnit.MILLISECONDS, DEFAULT_THRESHOLD);
+ }
+
+ public DownInstancePolicy(long timeout, TimeUnit unit, int threshold)
+ {
+ this.timeoutMs = unit.toMillis(timeout);
+ this.threshold = threshold;
+ }
+
+ public long getTimeoutMs()
+ {
+ return timeoutMs;
+ }
+
+ public int getThreshold()
+ {
+ return threshold;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-curator/blob/46d112f8/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProvider.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProvider.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProvider.java
index 3cdc0b5..44ee291 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProvider.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProvider.java
@@ -16,6 +16,7 @@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.curator.x.discovery;
import org.apache.curator.x.discovery.details.InstanceProvider;
@@ -41,5 +42,7 @@ public interface ServiceProvider<T> extends Closeable
* @return the instance to use
* @throws Exception any errors
*/
- public ServiceInstance<T> getInstance() throws Exception;
+ public ServiceInstance<T> getInstance() throws Exception;
+
+ public void noteError(ServiceInstance<T> instance);
}
http://git-wip-us.apache.org/repos/asf/incubator-curator/blob/46d112f8/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProviderBuilder.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProviderBuilder.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProviderBuilder.java
index e6eaa39..4a0e636 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProviderBuilder.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/ServiceProviderBuilder.java
@@ -55,7 +55,7 @@ public interface ServiceProviderBuilder<T>
*/
public ServiceProviderBuilder<T> threadFactory(ThreadFactory threadFactory);
- public ServiceProviderBuilder<T> downInstanceArguments(long timeout, TimeUnit unit, int threshold);
+ public ServiceProviderBuilder<T> downInstancePolicy(DownInstancePolicy downInstancePolicy);
public ServiceProviderBuilder<T> additionalFilter(InstanceFilter<T> filter);
}
http://git-wip-us.apache.org/repos/asf/incubator-curator/blob/46d112f8/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/DownInstanceManager.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/DownInstanceManager.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/DownInstanceManager.java
index efda05c..a70f205 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/DownInstanceManager.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/DownInstanceManager.java
@@ -19,21 +19,17 @@
package org.apache.curator.x.discovery.details;
import com.google.common.collect.Maps;
+import org.apache.curator.x.discovery.DownInstancePolicy;
import org.apache.curator.x.discovery.InstanceFilter;
import org.apache.curator.x.discovery.ServiceInstance;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
class DownInstanceManager<T> implements InstanceFilter<T>
{
- private final long timeoutMs;
- private final int threshold;
private final ConcurrentMap<ServiceInstance<?>, Status> statuses = Maps.newConcurrentMap();
-
- private static final long DEFAULT_TIMEOUT_MS = 30000;
- private static final int DEFAULT_THRESHOLD = 2;
+ private final DownInstancePolicy downInstancePolicy;
private static class Status
{
@@ -43,24 +39,14 @@ class DownInstanceManager<T> implements InstanceFilter<T>
DownInstanceManager()
{
- this(DEFAULT_TIMEOUT_MS, TimeUnit.MILLISECONDS, DEFAULT_THRESHOLD);
+ this(new DownInstancePolicy());
}
- /**
- * @param timeout amount of time for instances to be unavailable
- * @param unit time unit
- */
- DownInstanceManager(long timeout, TimeUnit unit, int threshold)
+ DownInstanceManager(DownInstancePolicy downInstancePolicy)
{
- this.threshold = threshold;
- timeoutMs = unit.toMillis(timeout);
+ this.downInstancePolicy = downInstancePolicy;
}
- /**
- * This instance will be unavailable for the configured timeout
- *
- * @param instance instance to mark unavailable
- */
void add(ServiceInstance<?> instance)
{
purge();
@@ -77,7 +63,7 @@ class DownInstanceManager<T> implements InstanceFilter<T>
purge();
Status status = statuses.get(instance);
- return (status == null) || (status.errorCount.get() < threshold);
+ return (status == null) || (status.errorCount.get() < downInstancePolicy.getThreshold());
}
private void purge()
@@ -85,7 +71,7 @@ class DownInstanceManager<T> implements InstanceFilter<T>
for ( Map.Entry<ServiceInstance<?>, Status> entry : statuses.entrySet() )
{
long elapsedMs = System.currentTimeMillis() - entry.getValue().startMs;
- if ( elapsedMs >= timeoutMs )
+ if ( elapsedMs >= downInstancePolicy.getTimeoutMs() )
{
statuses.remove(entry.getKey());
}
http://git-wip-us.apache.org/repos/asf/incubator-curator/blob/46d112f8/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderBuilderImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderBuilderImpl.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderBuilderImpl.java
index 7c811b5..5c020c5 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderBuilderImpl.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderBuilderImpl.java
@@ -19,6 +19,7 @@
package org.apache.curator.x.discovery.details;
import com.google.common.collect.Lists;
+import org.apache.curator.x.discovery.DownInstancePolicy;
import org.apache.curator.x.discovery.InstanceFilter;
import org.apache.curator.x.discovery.ProviderStrategy;
import org.apache.curator.x.discovery.ServiceProvider;
@@ -38,14 +39,12 @@ class ServiceProviderBuilderImpl<T> implements ServiceProviderBuilder<T>
private String serviceName;
private ProviderStrategy<T> providerStrategy;
private ThreadFactory threadFactory;
- private DownInstanceManager<T> downInstanceManager = new DownInstanceManager<T>();
private List<InstanceFilter<T>> filters = Lists.newArrayList();
+ private DownInstancePolicy downInstancePolicy = new DownInstancePolicy();
public ServiceProvider<T> build()
{
- ArrayList<InstanceFilter<T>> localFilters = Lists.newArrayList(filters);
- localFilters.add(downInstanceManager);
- return new ServiceProviderImpl<T>(discovery, serviceName, providerStrategy, threadFactory, filters);
+ return new ServiceProviderImpl<T>(discovery, serviceName, providerStrategy, threadFactory, filters, downInstancePolicy);
}
ServiceProviderBuilderImpl(ServiceDiscoveryImpl<T> discovery)
@@ -93,9 +92,9 @@ class ServiceProviderBuilderImpl<T> implements ServiceProviderBuilder<T>
}
@Override
- public ServiceProviderBuilder<T> downInstanceArguments(long timeout, TimeUnit unit, int threshold)
+ public ServiceProviderBuilder<T> downInstancePolicy(DownInstancePolicy downInstancePolicy)
{
- downInstanceManager = new DownInstanceManager<T>(timeout, unit, threshold);
+ this.downInstancePolicy = downInstancePolicy;
return this;
}
http://git-wip-us.apache.org/repos/asf/incubator-curator/blob/46d112f8/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderImpl.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderImpl.java b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderImpl.java
index dac5890..40dcf59 100644
--- a/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderImpl.java
+++ b/curator-x-discovery/src/main/java/org/apache/curator/x/discovery/details/ServiceProviderImpl.java
@@ -18,12 +18,15 @@
*/
package org.apache.curator.x.discovery.details;
+import com.google.common.collect.Lists;
+import org.apache.curator.x.discovery.DownInstancePolicy;
import org.apache.curator.x.discovery.InstanceFilter;
import org.apache.curator.x.discovery.ProviderStrategy;
import org.apache.curator.x.discovery.ServiceCache;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.ServiceProvider;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadFactory;
@@ -37,13 +40,19 @@ public class ServiceProviderImpl<T> implements ServiceProvider<T>
private final InstanceProvider<T> instanceProvider;
private final ServiceDiscoveryImpl<T> discovery;
private final ProviderStrategy<T> providerStrategy;
+ private final DownInstanceManager<T> downInstanceManager;
- public ServiceProviderImpl(ServiceDiscoveryImpl<T> discovery, String serviceName, ProviderStrategy<T> providerStrategy, ThreadFactory threadFactory, List<InstanceFilter<T>> filters)
+ public ServiceProviderImpl(ServiceDiscoveryImpl<T> discovery, String serviceName, ProviderStrategy<T> providerStrategy, ThreadFactory threadFactory, List<InstanceFilter<T>> filters, DownInstancePolicy downInstancePolicy)
{
this.discovery = discovery;
this.providerStrategy = providerStrategy;
+
+ downInstanceManager = new DownInstanceManager<T>(downInstancePolicy);
cache = discovery.serviceCacheBuilder().name(serviceName).threadFactory(threadFactory).build();
- instanceProvider = new FilteredInstanceProvider<T>(cache, filters);
+
+ ArrayList<InstanceFilter<T>> localFilters = Lists.newArrayList(filters);
+ localFilters.add(downInstanceManager);
+ instanceProvider = new FilteredInstanceProvider<T>(cache, localFilters);
}
/**
@@ -80,4 +89,10 @@ public class ServiceProviderImpl<T> implements ServiceProvider<T>
{
return providerStrategy.getInstance(instanceProvider);
}
+
+ @Override
+ public void noteError(ServiceInstance<T> instance)
+ {
+ downInstanceManager.add(instance);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-curator/blob/46d112f8/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/details/TestDownInstanceManager.java
----------------------------------------------------------------------
diff --git a/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/details/TestDownInstanceManager.java b/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/details/TestDownInstanceManager.java
index bddae89..ff4d4f8 100644
--- a/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/details/TestDownInstanceManager.java
+++ b/curator-x-discovery/src/test/java/org/apache/curator/x/discovery/details/TestDownInstanceManager.java
@@ -31,7 +31,7 @@ public class TestDownInstanceManager
ServiceInstance<Void> instance1 = ServiceInstance.<Void>builder().name("hey").id("1").build();
ServiceInstance<Void> instance2 = ServiceInstance.<Void>builder().name("hey").id("2").build();
- DownInstanceManager<Void> downInstanceManager = new DownInstanceManager<Void>(10, TimeUnit.DAYS, 1);
+ DownInstanceManager<Void> downInstanceManager = new DownInstanceManager<Void>();
Assert.assertTrue(downInstanceManager.apply(instance1));
Assert.assertTrue(downInstanceManager.apply(instance2));
@@ -46,7 +46,7 @@ public class TestDownInstanceManager
ServiceInstance<Void> instance1 = ServiceInstance.<Void>builder().name("hey").id("1").build();
ServiceInstance<Void> instance2 = ServiceInstance.<Void>builder().name("hey").id("2").build();
- DownInstanceManager<Void> downInstanceManager = new DownInstanceManager<Void>(1, TimeUnit.SECONDS, 1);
+ DownInstanceManager<Void> downInstanceManager = new DownInstanceManager<Void>();
downInstanceManager.add(instance1);
Assert.assertFalse(downInstanceManager.apply(instance1));
@@ -64,7 +64,7 @@ public class TestDownInstanceManager
ServiceInstance<Void> instance1 = ServiceInstance.<Void>builder().name("hey").id("1").build();
ServiceInstance<Void> instance2 = ServiceInstance.<Void>builder().name("hey").id("2").build();
- DownInstanceManager<Void> downInstanceManager = new DownInstanceManager<Void>(1, TimeUnit.SECONDS, 1);
+ DownInstanceManager<Void> downInstanceManager = new DownInstanceManager<Void>();
ServiceDiscoveryImpl<Void> discovery = new ServiceDiscoveryImpl<Void>(null, null, null, null);
}
}