You are viewing a plain text version of this content. The canonical link for it is here.
Posted to s4-commits@incubator.apache.org by mm...@apache.org on 2012/01/03 11:19:16 UTC
[47/50] [abbrv] git commit: Add AbstractPE.expire()
Add AbstractPE.expire()
Project: http://git-wip-us.apache.org/repos/asf/incubator-s4/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-s4/commit/b11da4f4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-s4/tree/b11da4f4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-s4/diff/b11da4f4
Branch: refs/heads/dev
Commit: b11da4f4ebf242d99626a8395f344ae41520ba70
Parents: 737fa28
Author: Bruce Robbins <ro...@thosespent.corp.yahoo.com>
Authored: Tue Oct 11 13:53:57 2011 -0700
Committer: Bruce Robbins <ro...@thosespent.corp.yahoo.com>
Committed: Tue Oct 11 13:53:57 2011 -0700
----------------------------------------------------------------------
.../main/java/io/s4/persist/ConMapPersister.java | 4 +++
.../main/java/io/s4/persist/HashMapPersister.java | 5 ++++
.../src/main/java/io/s4/processor/AbstractPE.java | 12 ++++++++++
.../java/io/s4/processor/PrototypeWrapper.java | 17 +++++++++-----
4 files changed, 32 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/b11da4f4/s4-core/src/main/java/io/s4/persist/ConMapPersister.java
----------------------------------------------------------------------
diff --git a/s4-core/src/main/java/io/s4/persist/ConMapPersister.java b/s4-core/src/main/java/io/s4/persist/ConMapPersister.java
index c47952f..4aa85ed 100644
--- a/s4-core/src/main/java/io/s4/persist/ConMapPersister.java
+++ b/s4-core/src/main/java/io/s4/persist/ConMapPersister.java
@@ -110,6 +110,10 @@ public class ConMapPersister implements Persister {
}
public void set(String key, Object value, int period) {
+ if (value == null) {
+ cache.remove(key);
+ return;
+ }
persistCount.getAndIncrement();
CacheEntry ce = new CacheEntry();
ce.value = value;
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/b11da4f4/s4-core/src/main/java/io/s4/persist/HashMapPersister.java
----------------------------------------------------------------------
diff --git a/s4-core/src/main/java/io/s4/persist/HashMapPersister.java b/s4-core/src/main/java/io/s4/persist/HashMapPersister.java
index f59d338..483feb1 100644
--- a/s4-core/src/main/java/io/s4/persist/HashMapPersister.java
+++ b/s4-core/src/main/java/io/s4/persist/HashMapPersister.java
@@ -108,6 +108,11 @@ public class HashMapPersister implements Persister {
}
public void set(String key, Object value, int period) {
+ if (value == null) {
+ cache.remove(key);
+ return;
+ }
+
synchronized (this) {
persistCount++;
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/b11da4f4/s4-core/src/main/java/io/s4/processor/AbstractPE.java
----------------------------------------------------------------------
diff --git a/s4-core/src/main/java/io/s4/processor/AbstractPE.java b/s4-core/src/main/java/io/s4/processor/AbstractPE.java
index e0a60f1..2f0480a 100644
--- a/s4-core/src/main/java/io/s4/processor/AbstractPE.java
+++ b/s4-core/src/main/java/io/s4/processor/AbstractPE.java
@@ -102,6 +102,7 @@ public abstract class AbstractPE implements Cloneable {
transient private boolean logPauses = false;
private String id;
transient protected SchemaContainer schemaContainer = new SchemaContainer();
+ transient private PrototypeWrapper prototypeWrapper;
transient private boolean recoveryAttempted = false;
// true if state may have changed
@@ -172,6 +173,10 @@ public abstract class AbstractPE implements Cloneable {
return clock;
}
+ public void setPrototypeWrapper(PrototypeWrapper prototypeWrapper) {
+ this.prototypeWrapper = prototypeWrapper;
+ }
+
public AbstractPE() {
OverloadDispatcherGenerator oldg = new OverloadDispatcherGenerator(this.getClass());
Class<?> overloadDispatcherClass = oldg.generate();
@@ -649,6 +654,13 @@ public abstract class AbstractPE implements Cloneable {
isCheckpointingEvent = true;
recover();
}
+
+ /**
+ * This method expires the current PE.
+ **/
+ protected void expire() {
+ this.prototypeWrapper.expire(this.keyValueString);
+ }
class PeriodicInvoker implements Runnable {
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/b11da4f4/s4-core/src/main/java/io/s4/processor/PrototypeWrapper.java
----------------------------------------------------------------------
diff --git a/s4-core/src/main/java/io/s4/processor/PrototypeWrapper.java b/s4-core/src/main/java/io/s4/processor/PrototypeWrapper.java
index 00e6f17..93a2d16 100644
--- a/s4-core/src/main/java/io/s4/processor/PrototypeWrapper.java
+++ b/s4-core/src/main/java/io/s4/processor/PrototypeWrapper.java
@@ -20,7 +20,6 @@ import io.s4.persist.ConMapPersister;
import io.s4.persist.Persister;
import io.s4.util.clock.Clock;
-import java.lang.reflect.Method;
import java.util.List;
import org.apache.log4j.Logger;
@@ -47,11 +46,8 @@ public class PrototypeWrapper {
((ConMapPersister) lookupTable).setSelfClean(true);
((ConMapPersister) lookupTable).init();
// set the persister in prototype
- Method method = prototype.getClass().getMethod("setLookupTable",
- Persister.class);
- method.invoke(prototype, lookupTable);
- } catch (NoSuchMethodException e) {
- // this is expected
+ prototype.setLookupTable(lookupTable);
+ prototype.setPrototypeWrapper(this);
} catch (Exception e) {
// this is not expected
Logger.getLogger("s4")
@@ -109,6 +105,15 @@ public class PrototypeWrapper {
return pe;
}
+
+ public void expire(String keyValue) {
+ try {
+ lookupTable.set(keyValue, null, 0);
+ } catch (Exception e) {
+ logger.error("exception when removing pe for key:" + keyValue, e);
+ }
+
+ }
public int getPECount() {
return lookupTable.keySet().size();