Changeset 243

Timestamp:

07/11/13 19:08:33 (10 years ago)

Author:

bowman

Message:

Completed ghttp test-backend (now supports /f/ operation) and tweaked CacheCoveragePlugin http forwarder to match.

Location:

trunk

Files:

• plugins/CacheCoveragePlugin/src/com/bowman/cardserv/HttpCacheForwarder.java (modified) (7 diffs)
• src/com/bowman/cardserv/CardServProxy.java (modified) (1 diff)
• src/com/bowman/cardserv/DefaultCache.java (modified) (3 diffs)
• src/com/bowman/cardserv/interfaces/CacheHandler.java (modified) (2 diffs)
• src/com/bowman/cardserv/web/GHttpBackend.java (modified) (8 diffs)

trunk/plugins/CacheCoveragePlugin/src/com/bowman/cardserv/HttpCacheForwarder.java

@@ -8 +8 @@
 
 import java.io.*;
+import java.lang.System;
 import java.net.*;
 import java.util.*;
@@ -37 +38 @@
   private int port;
   private boolean connected, redundant;
-  private int counter, reconnects, errors, ecmForwards, delayAlerts, filtered;
+  private int counter, reconnects, timeouts, errors, ecmForwards, delayAlerts, filtered;
   private int maxDelay;
   private Set profiles, caids;
@@ -105 +106 @@
     p.setProperty("reconnects", String.valueOf(reconnects));
     p.setProperty("errors", String.valueOf(errors));
+    p.setProperty("timeouts", String.valueOf(timeouts));
     return p;
   }
@@ -223 +225 @@
         headers.setProperty(s[0], s[1]);
         if(s[0].equals("Set-Cookie")) {
-          s = s[1].split("=");
+          if(s[1].indexOf(";") != -1) {
+            s = s[1].split(";"); // ignore all except 1st
+            s = s[0].split("=");
+          } else s = s[1].split("=");
           cookies.setProperty(s[0], s[1]);
         }
@@ -345 +350 @@
         reply.body = sb.toString().toCharArray();
       } else {
-        reply.body = new char[reply.getContentLength()];
-        if(br.read(reply.body) != reply.body.length) {
-          throw new IOException("Assertation failed"); // todo
-        }
+        if(reply.getContentLength() > 0) {
+          reply.body = new char[reply.getContentLength()];
+          if(br.read(reply.body) != reply.body.length) {
+            throw new IOException("Assertation failed"); // todo
+          }
+        } else reply.body = new char[0];
       }
 
@@ -488 +495 @@
               int curSize = redundant?localQ.size():singleQ.size();
               if(maxDelay > 0 && curSize < 100) Thread.sleep(maxDelay);
-            } catch (SocketException se) {
+            } catch(SocketException se) {
               // socket was closed gracefully, reconnect immediately
               parent.logger.throwing(se);
@@ -496 +503 @@
           }
         } catch(SocketException e) { // probably connect failure?
-          parent.logger.info("HttpCacheForwarder[" + name + "] disconnected");
+          parent.logger.info("HttpCacheForwarder[" + name + "] failed to connect: " + e);
           parent.logger.throwing(e);
+          handleError(myQ, false);
+        } catch(SocketTimeoutException e) { // timeout
+          parent.logger.warning("HttpCacheForwarder[" + name + "] timeout (sendQ size: " + myQ.size() + "): " + e);
+          parent.logger.throwing(e);
+          timeouts++;
           handleError(myQ, false);
         } catch(IOException e) { // abnormal disconnect

trunk/src/com/bowman/cardserv/CardServProxy.java

@@ -401 +401 @@
 
         if(modified.isEmpty() && !(profile.isCacheOnly() || config.isCatchAll())) {
-          logger.fine("Denying message with no connector candidates from '" + session + "': " + msg.hashCodeStr());
-          denyMessage(session, msg);
-          return;
+          if(!cacheHandler.containsCaid(msg.getCaId())) {
+            logger.fine("Denying message with no connector candidates from '" + session + "': " + msg.hashCodeStr());
+            denyMessage(session, msg);
+            return;
+          }
         }
 

trunk/src/com/bowman/cardserv/DefaultCache.java

@@ -24 +24 @@
 
   private int timeouts, instantHits, waitHits, remoteHits, pendingPeak, contested;
+
+  private Set caids = new HashSet();
 
   public DefaultCache() {
@@ -162 +164 @@
   }
 
+  public synchronized void processReplies(Map replies) {
+    boolean lockFound = false;
+    CamdNetMessage request, reply;
+    for(Iterator iter = replies.keySet().iterator(); iter.hasNext(); ) {
+      request = (CamdNetMessage)iter.next();
+      reply = (CamdNetMessage)replies.get(request);
+      // if(monitor != null) monitor.onReply(request, reply);
+      if(listener != null) listener.onReply(request, reply);
+      if(pendingEcms.containsKey(request)) {
+        removeRequest(request);
+        lockFound = true;
+      }
+      addReply(request, reply);
+    }
+    if(lockFound) notifyAll();
+  }
+
   public CamdNetMessage peekReply(CamdNetMessage request) {
     return (CamdNetMessage)ecmMap.get(request);
@@ -186 +205 @@
   }
 
+  public boolean containsCaid(int caid) {
+    return caids.contains(new Integer(caid));
+  }
+
   protected synchronized void addRequest(int successFactor, CamdNetMessage request, boolean alwaysWait) {
     CamdNetMessage oldRequest = (CamdNetMessage)pendingEcms.put(request, request);
     if(pendingEcms.size() > pendingPeak) pendingPeak = pendingEcms.size();
     if(oldRequest == null) {
+      caids.add(new Integer(request.getCaId()));
       if(monitor != null) monitor.onRequest(successFactor, request);
       if(listener != null) listener.onRequest(successFactor, request);

trunk/src/com/bowman/cardserv/interfaces/CacheHandler.java

@@ -3 +3 @@
 import com.bowman.cardserv.CamdNetMessage;
 
-import java.util.Properties;
+import java.util.*;
 
 /**
@@ -18 +18 @@
   CamdNetMessage peekReply(CamdNetMessage request);
   long getMaxCacheWait(long maxCwWait);
+  boolean containsCaid(int caid);
+  void processReplies(Map replies);
 
   Properties getUsageStats();

trunk/src/com/bowman/cardserv/web/GHttpBackend.java

@@ -40 +40 @@
   private Map sessionsByUser = new HashMap();
   private Map contexts = Collections.synchronizedMap(new LinkedHashMap());
+  private Set blackList = new HashSet();
 
   private StatusCommand contextsCmd;
@@ -144 +145 @@
       }
 
+      blackList.clear();
+      String bl = FileFetcher.getProperty("cache.bl");
+      if(bl != null) blackList.addAll(Arrays.asList(bl.split(" ")));
+
     } else {
       if(httpd != null) {
@@ -191 +196 @@
 
   private CamdNetMessage waitForReply(CamdNetMessage request) {
+    // long start = System.currentTimeMillis();
     CamdNetMessage reply = cache.peekReply(request);
     if(reply != null) return reply;
-    else reply = cache.processRequest(-1, request, true, request.getMaxWait() * 3); // todo
+    else reply = cache.processRequest(-1, request, true, request.getMaxWait() * 2); // todo
+    // System.out.println((System.currentTimeMillis() - start) + " - " + reply);
     return reply;
   }
@@ -211 +218 @@
   }
 
-  private GHttpSession getSession(String id, String ip) {
+  private GHttpSession getSession(String id, String ip) throws GHttpAuthException {
     GHttpSession gs = (GHttpSession)sessions.get(id);
     if(gs != null && gs.isExpired()) {
       sessionsByUser.remove(gs.getUser());
       sessions.remove(id);
-      gs = null;
+      throw new GHttpAuthException(getErrorResponse(401, "Authorization required"));
     }
     if(gs != null && ip.equals(gs.getRemoteAddress())) return gs;
@@ -222 +229 @@
   }
 
-  private GHttpSession findSession(String user, String ip) {
+  private GHttpSession findSession(String user, String ip) throws GHttpAuthException {
     GHttpSession gs = (GHttpSession)sessionsByUser.get(user);
     if(gs != null && gs.isExpired()) {
       sessionsByUser.remove(user);
       sessions.remove(gs.getGhttpSessionId());
-      gs = null;
+      throw new GHttpAuthException(getErrorResponse(401, "Authorization required"));
     }
     if(gs != null && ip.equals(gs.getRemoteAddress())) return gs;
@@ -234 +241 @@
 
   private synchronized GHttpSession doGhttpAuth(HttpRequest req, String[] s) throws GHttpAuthException {
+    return doGhttpAuth(req, s, accessPasswd);
+  }
+
+  private synchronized GHttpSession doGhttpAuth(HttpRequest req, String[] s, String passwd) throws GHttpAuthException {
     GHttpSession gs = null;
-    if(s[3].length() >= 6) { // session id included
+    if(s.length > 3 && s[3].length() >= 6) { // session id included
       gs = getSession(s[3], req.getRemoteAddress());
     }
     if(gs == null) { // no session id in query string (or invalid/expired session), check for auth
-      String user = parent.checkBasicAuth(req, accessPasswd);
-      if(user == null) throw new GHttpAuthException(getErrorResponse(401, "Authorization required"));
+      String user = parent.checkBasicAuth(req, passwd);
+      if(user == null) throw new GHttpAuthException(getErrorResponse(403, "Forbidden"));
       if(!user.startsWith(WebBackend.anonPrefix)) {
         HttpResponse error = parent.doIpCheck(req, user);
@@ -298 +309 @@
          return doPmtGetOrPost(req);
       } else if(q.startsWith(API_FEEDER_POST)) {
-        // todo
+        return doCachePost(req);
       }
       return getErrorResponse(503, "Not implemented"); // todo
+    }
+  }
+
+  private HttpResponse doCachePost(HttpRequest req) {
+    if(feederPasswd == null) return getErrorResponse(403, "Disabled");
+    if(cache == null) cache = config.getCacheHandler();
+    try {
+      String[] s = req.getQueryString().split("/");
+      GHttpSession gs = doGhttpAuth(req, s, feederPasswd);
+      if(gs == null) return getErrorResponse(403, "Forbidden");
+      byte[] content = req.getContent();
+      String addr = req.getRemoteAddress();
+      if(content.length >= RECORD_SIZE) {
+        Map replies = new HashMap();
+        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(content));
+        while(dis.available() >= RECORD_SIZE) {
+          CamdNetMessage request = CamdNetMessage.parseCacheReq(dis, false);
+          CamdNetMessage reply = CamdNetMessage.parseCacheRpl(dis, request, false);
+          if(!blackList.contains(addr)) {
+            request.setOriginAddress(addr);
+            reply.setOriginAddress(addr);
+          }
+          replies.put(request, reply);
+        }
+        cache.processReplies(replies);
+        if(dis.available() != 0) parent.logger.warning("Trailing bytes in feed post: " + dis.available());
+        HttpResponse res = new HttpResponse(204, "No Content"); // todo: include stats
+        if(s.length <= 3) res.setCookie("GSSID", gs.getGhttpSessionId());
+        return res;
+      } else throw new IOException("Invalid content length (" + content.length + ")");
+    } catch (GHttpAuthException e) {
+      return getErrorResponse(403, "Forbidden");
+    } catch (Exception e) {
+      parent.logger.throwing("Bad ghttp feeder request: " + req.getQueryString(), e);
+      return HttpResponse.getErrorResponse(400, req.getQueryString());
     }
   }
@@ -317 +363 @@
         if(ProxyConfig.getInstance().getProfileById(ecmReq.getNetworkId(), ecmReq.getCaId()) == null)
           return getErrorResponse(503, "Unknown system: " + CaProfile.getKeyStr(ecmReq.getNetworkId(), ecmReq.getCaId()));
+        else if(!cache.containsCaid(ecmReq.getCaId()))
+          return getErrorResponse(503, "Unknown caid: " + CaProfile.getKeyStr(ecmReq.getNetworkId(), ecmReq.getCaId()));
         gs.fireCamdMessage(ecmReq, false);
         reply = gs.waitForReply(ecmReq);
       } else instant = true;
 
-      if(reply == null) cache.peekReply(ecmReq);
+      if(reply == null) reply = cache.peekReply(ecmReq);
       if(reply == null) return getErrorResponse(503, "Ecm timeout");
       else {