curator-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From randg...@apache.org
Subject git commit: expiration should now be on a pre-object basis
Date Sun, 16 Feb 2014 00:39:45 GMT
Repository: curator
Updated Branches:
  refs/heads/CURATOR-88 8f6edd706 -> ce456d2df


expiration should now be on a pre-object basis


Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/ce456d2d
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/ce456d2d
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/ce456d2d

Branch: refs/heads/CURATOR-88
Commit: ce456d2df24c9c64108d1c40e111b66a7e00e3ca
Parents: 8f6edd7
Author: randgalt <randgalt@apache.org>
Authored: Sat Feb 15 19:39:38 2014 -0500
Committer: randgalt <randgalt@apache.org>
Committed: Sat Feb 15 19:39:38 2014 -0500

----------------------------------------------------------------------
 .../curator/x/rest/CuratorRestContext.java      | 38 +-------------
 .../curator/x/rest/api/ClientResource.java      | 24 ++++-----
 .../apache/curator/x/rest/api/Constants.java    |  2 +-
 .../org/apache/curator/x/rest/api/Session.java  | 55 +++++++++++++-------
 4 files changed, 50 insertions(+), 69 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/curator/blob/ce456d2d/curator-x-rest/src/main/java/org/apache/curator/x/rest/CuratorRestContext.java
----------------------------------------------------------------------
diff --git a/curator-x-rest/src/main/java/org/apache/curator/x/rest/CuratorRestContext.java
b/curator-x-rest/src/main/java/org/apache/curator/x/rest/CuratorRestContext.java
index cca2e89..a07197e 100644
--- a/curator-x-rest/src/main/java/org/apache/curator/x/rest/CuratorRestContext.java
+++ b/curator-x-rest/src/main/java/org/apache/curator/x/rest/CuratorRestContext.java
@@ -20,14 +20,10 @@ package org.apache.curator.x.rest;
 
 import com.google.common.base.Preconditions;
 import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.state.ConnectionState;
-import org.apache.curator.framework.state.ConnectionStateListener;
 import org.apache.curator.utils.ThreadUtils;
 import org.apache.curator.x.rest.api.Session;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.map.ObjectWriter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import java.io.Closeable;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -35,7 +31,6 @@ import java.util.concurrent.atomic.AtomicReference;
 
 public class CuratorRestContext implements Closeable
 {
-    private final Logger log = LoggerFactory.getLogger(getClass());
     private final Session session = new Session();
     private final ObjectMapper mapper = new ObjectMapper();
     private final ObjectWriter writer = mapper.writer();
@@ -43,17 +38,6 @@ public class CuratorRestContext implements Closeable
     private final int sessionLengthMs;
     private final AtomicReference<State> state = new AtomicReference<State>(State.LATENT);
     private final ScheduledExecutorService executorService = ThreadUtils.newSingleThreadScheduledExecutor("CuratorRestContext");
-    private final ConnectionStateListener connectionStateListener = new ConnectionStateListener()
-    {
-        @Override
-        public void stateChanged(CuratorFramework client, ConnectionState newState)
-        {
-            if ( newState == ConnectionState.LOST )
-            {
-                handleLostConnection();
-            }
-        }
-    };
 
     private enum State
     {
@@ -77,7 +61,6 @@ public class CuratorRestContext implements Closeable
     public Session getSession()
     {
         Preconditions.checkState(state.get() == State.STARTED, "Not started");
-        session.updateLastUse();
         return session;
     }
 
@@ -85,35 +68,22 @@ public class CuratorRestContext implements Closeable
     {
         Preconditions.checkState(state.compareAndSet(State.LATENT, State.STARTED), "Already
started");
 
-        client.getConnectionStateListenable().addListener(connectionStateListener);
-
         Runnable runner = new Runnable()
         {
             @Override
             public void run()
             {
-                checkSession();
+                session.checkExpiredThings(sessionLengthMs);
             }
         };
         executorService.scheduleAtFixedRate(runner, sessionLengthMs, sessionLengthMs, TimeUnit.MILLISECONDS);
     }
 
-    private void checkSession()
-    {
-        long elapsedSinceLastUse = System.currentTimeMillis() - session.getLastUseMs();
-        if ( elapsedSinceLastUse > sessionLengthMs )
-        {
-            log.warn("Session has expired. Closing all open recipes. Milliseconds since last
ping: " + elapsedSinceLastUse);
-            session.closeThings();
-        }
-    }
-
     @Override
     public void close()
     {
         if ( state.compareAndSet(State.STARTED, State.CLOSED) )
         {
-            client.getConnectionStateListenable().removeListener(connectionStateListener);
             executorService.shutdownNow();
             session.close();
         }
@@ -128,10 +98,4 @@ public class CuratorRestContext implements Closeable
     {
         return writer;
     }
-
-    private void handleLostConnection()
-    {
-        log.warn("Connection lost - closing all REST sessions");
-        session.closeThings();
-    }
 }

http://git-wip-us.apache.org/repos/asf/curator/blob/ce456d2d/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/ClientResource.java
----------------------------------------------------------------------
diff --git a/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/ClientResource.java
b/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/ClientResource.java
index 94009f0..26708cf 100644
--- a/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/ClientResource.java
+++ b/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/ClientResource.java
@@ -34,6 +34,7 @@ import javax.ws.rs.GET;
 import javax.ws.rs.POST;
 import javax.ws.rs.PUT;
 import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.WebApplicationException;
 import javax.ws.rs.core.Context;
@@ -64,14 +65,23 @@ public class ClientResource
         return Response.ok(context.getWriter().writeValueAsString(node)).build();
     }
 
+    @GET
+    @Path("/touch/{id}")
+    public Response touchThing(@PathParam("id") String id)
+    {
+        if ( !context.getSession().updateThingLastUse(id) )
+        {
+            return Response.status(Response.Status.NOT_FOUND).build();
+        }
+        return Response.ok().build();
+    }
+
     @POST
     @Consumes(MediaType.APPLICATION_JSON)
     @Produces(MediaType.APPLICATION_JSON)
     @Path("/get-children")
     public Response getChildren(final GetChildrenSpec getChildrenSpec) throws Exception
     {
-        context.getSession();   // update last use
-
         Object builder = context.getClient().getChildren();
         if ( getChildrenSpec.isWatched() )
         {
@@ -110,8 +120,6 @@ public class ClientResource
     @Path("/delete")
     public Response delete(final DeleteSpec deleteSpec) throws Exception
     {
-        context.getSession();   // update last use
-
         Object builder = context.getClient().delete();
         if ( deleteSpec.isGuaranteed() )
         {
@@ -135,8 +143,6 @@ public class ClientResource
     @Path("/set-data")
     public Response setData(final SetDataSpec setDataSpec) throws Exception
     {
-        context.getSession();   // update last use
-
         Object builder = context.getClient().setData();
         if ( setDataSpec.isCompressed() )
         {
@@ -164,8 +170,6 @@ public class ClientResource
     @Path("/create")
     public Response create(final CreateSpec createSpec) throws Exception
     {
-        context.getSession();   // update last use
-
         Object builder = context.getClient().create();
         if ( createSpec.isCreatingParentsIfNeeded() )
         {
@@ -200,8 +204,6 @@ public class ClientResource
     @Path("/get-data")
     public Response getData(final GetDataSpec getDataSpec) throws Exception
     {
-        context.getSession();   // update last use
-
         Object builder = context.getClient().getData();
         if ( getDataSpec.isWatched() )
         {
@@ -243,8 +245,6 @@ public class ClientResource
     @Path("/exists")
     public Response exists(final ExistsSpec existsSpec) throws Exception
     {
-        context.getSession();   // update last use
-
         Object builder = context.getClient().checkExists();
         if ( existsSpec.isWatched() )
         {

http://git-wip-us.apache.org/repos/asf/curator/blob/ce456d2d/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Constants.java
----------------------------------------------------------------------
diff --git a/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Constants.java b/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Constants.java
index b12d1b1..5ee1982 100644
--- a/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Constants.java
+++ b/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Constants.java
@@ -38,7 +38,7 @@ class Constants
     static final String PATH_CACHE = "path-cache";
     static final String NODE_CACHE = "node-cache";
     static final String LEADER = "leader";
-    static final String CLOSING = "closing";
+    static final String EXPIRED = "expired";
 
     static ObjectNode makeIdNode(CuratorRestContext context, String id)
     {

http://git-wip-us.apache.org/repos/asf/curator/blob/ce456d2d/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Session.java
----------------------------------------------------------------------
diff --git a/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Session.java b/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Session.java
index 2d2cd8a..16478ce 100644
--- a/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Session.java
+++ b/curator-x-rest/src/main/java/org/apache/curator/x/rest/api/Session.java
@@ -35,13 +35,13 @@ public class Session implements Closeable
 {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final Map<String, Entry> things = Maps.newConcurrentMap();
-    private final AtomicLong lastUseMs = new AtomicLong(System.currentTimeMillis());
     private final BlockingQueue<StatusMessage> messages = Queues.newLinkedBlockingQueue();
 
     private static class Entry
     {
         final Object thing;
         final Closer closer;
+        final AtomicLong lastUseMs = new AtomicLong(System.currentTimeMillis());
 
         private Entry(Object thing, Closer closer)
         {
@@ -50,26 +50,9 @@ public class Session implements Closeable
         }
     }
 
-    public void updateLastUse()
-    {
-        lastUseMs.set(System.currentTimeMillis());
-    }
-
-    public long getLastUseMs()
-    {
-        return lastUseMs.get();
-    }
-
     @Override
-    public void close()
+    public synchronized void close()
     {
-        closeThings();
-    }
-
-    public void closeThings()
-    {
-        pushMessage(new StatusMessage(Constants.CLOSING, "", "", ""));
-        
         for ( Map.Entry<String, Entry> mapEntry : things.entrySet() )
         {
             Entry entry = mapEntry.getValue();
@@ -80,6 +63,29 @@ public class Session implements Closeable
                 entry.closer.close(entry.thing);    // lack of generics is safe because addThing()
is type-safe
             }
         }
+        things.clear();
+    }
+
+    public synchronized void checkExpiredThings(long sessionLengthMs)
+    {
+        for ( Map.Entry<String, Entry> mapEntry : things.entrySet() )
+        {
+            Entry entry = mapEntry.getValue();
+            long elapsedSinceLastUse = System.currentTimeMillis() - entry.lastUseMs.get();
+            if ( elapsedSinceLastUse > sessionLengthMs )
+            {
+                String id = mapEntry.getKey();
+                pushMessage(new StatusMessage(Constants.EXPIRED, id, "expired", entry.thing.getClass().getName()));
+                log.warn(String.format("Expiring object. Elapsed time: %d, id: %s, Class:
%s", elapsedSinceLastUse, id, entry.thing.getClass().getName()));
+
+                things.remove(id);
+                if ( entry.closer != null )
+                {
+                    //noinspection unchecked
+                    entry.closer.close(entry.thing);    // lack of generics is safe because
addThing() is type-safe
+                }
+            }
+        }
     }
 
     void pushMessage(StatusMessage message)
@@ -94,6 +100,17 @@ public class Session implements Closeable
         return localMessages;
     }
 
+    boolean updateThingLastUse(String id)
+    {
+        Entry entry = things.get(id);
+        if ( entry != null )
+        {
+            entry.lastUseMs.set(System.currentTimeMillis());
+            return true;
+        }
+        return false;
+    }
+
     <T> String addThing(T thing, Closer<T> closer)
     {
         String id = Constants.newId();


Mime
View raw message