new buckets

Author: robbins

java/src/com/ibm/javametrics/agent/Bucket.java

@@ -18,7 +18,9 @@
 public interface Bucket {
     public void pushData(String data);
 
-    public String empty();
+    public String getNext();
+    
+    public void empty();
 
     public int getSize();
 

java/src/com/ibm/javametrics/agent/JavaAgent.java

@@ -44,33 +44,41 @@ public JavaAgent() {
 
     private void init() {
         exec = Executors.newSingleThreadScheduledExecutor();
-        exec.scheduleAtFixedRate(this::empty, collectionInterval, collectionInterval, TimeUnit.SECONDS);
+        exec.scheduleAtFixedRate(this::drain, collectionInterval, collectionInterval, TimeUnit.SECONDS);
     }
 
     public void pushData(String type, String data) {
         synchronized (buckets) {
             Bucket bucket = buckets.get(type);
             if (bucket == null) {
-                bucket = new StringDataBucket();
+                bucket = new JsonDataBucket();
                 buckets.put(type, bucket);
             }
 
             if ((bucket.getSize() + data.length()) > MAX_BUCKET_SIZE) {
-                emit(type, bucket.empty());
+                drain(type, bucket);
             }
 
             bucket.pushData(data);
         }
     }
 
-    private void empty() {
+    private void drain() {
         synchronized (buckets) {
             buckets.forEach((name, bucket) -> {
-                emit(name, bucket.empty());
+                drain(name, bucket);
             });
         }
     }
 
+    private void drain(String type, Bucket bucket) {
+        String data = bucket.getNext();
+        while (data != null) {
+            emit(type, data);
+            data = bucket.getNext();
+        }
+    }
+    
     private void emit(String type, String data) {
         if (data != null) {
             receivers.forEach((receiver) -> {

java/src/com/ibm/javametrics/agent/JsonDataBucket.java

@@ -0,0 +1,69 @@
+/*******************************************************************************
+ * Copyright 2017 IBM Corp.
+ *
+ * Licensed 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 com.ibm.javametrics.agent;
+
+import java.util.ArrayList;
+
+public class JsonDataBucket implements Bucket {
+
+    private static final int MAX_SIZE = 4 * 1024;
+    private ArrayList<String> bucket = new ArrayList<String>(100);
+    int size = 0;
+    int cursor = 0;
+
+    @Override
+    public int getSize() {
+        return size;
+    }
+
+    @Override
+    public void empty() {
+        synchronized (bucket) {
+            bucket.clear();
+            cursor = 0;
+            size = 0;
+        }
+    }
+
+    @Override
+    public void pushData(String data) {
+        synchronized (bucket) {
+            bucket.add(data);
+            size += data.length();
+
+            System.err.println("pushed " + cursor + " " + bucket.size() + " " + size);
+            while (size > MAX_SIZE && cursor > 0) {
+                String removed = bucket.remove(0);
+                cursor--;
+                size -= removed.length();
+                System.err.println("trimmed " + cursor + " " + bucket.size() + " " + size);
+            }
+        }
+    }
+
+    @Override
+    public String getNext() {
+        String data = null;
+        synchronized (bucket) {
+            if (cursor < bucket.size()) {
+                data = bucket.get(cursor);
+                cursor++;
+                System.err.println("returned " + cursor + " " + bucket.size() + " " + size);
+            }
+        }
+        return data;
+    }
+}

java/src/com/ibm/javametrics/agent/StringDataBucket.java

@@ -26,14 +26,19 @@ public int getSize() {
     }
 
     @Override
-    public String empty() {
-        String data = bucket.toString();
+    public void empty() {
         bucket = new StringBuffer(INITIAL_BUCKET_SIZE);
-        return data;
     }
 
     @Override
     public void pushData(String data) {
         bucket.append(data);
     }
+
+    @Override
+    public String getNext() {
+        String data = bucket.toString();
+        bucket = new StringBuffer(INITIAL_BUCKET_SIZE);
+        return data;
+    }
 }

java/src/com/ibm/javametrics/impl/JavametricsImpl.java

@@ -51,7 +51,7 @@ public JavametricsImpl() {
 
     private void initializeProviders() {
         /*
-         * Start the mbean data providers
+         * Start the data providers
          */
         providerManager = new DataProviderManager(COLLECTION_INTERVAL);
     }
@@ -63,6 +63,7 @@ private void initializeProviders() {
      * java.lang.String)
      */
     public void receiveData(String type, String data) {
+        System.err.println("received: " + type + " " + data);
         for (Iterator<JavametricsListener> iterator = javametricsListeners.iterator(); iterator.hasNext();) {
             JavametricsListener javametricsListener = iterator.next();
             javametricsListener.receive(type, data);