metron-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ceste...@apache.org
Subject [52/66] [abbrv] incubator-metron git commit: METRON-239: NOOP Bulk Message Writer. This closes apache/incubator-metron#162
Date Fri, 24 Jun 2016 20:57:50 GMT
METRON-239: NOOP Bulk Message Writer. This closes apache/incubator-metron#162


Project: http://git-wip-us.apache.org/repos/asf/incubator-metron/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-metron/commit/72559d1e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-metron/tree/72559d1e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-metron/diff/72559d1e

Branch: refs/heads/master
Commit: 72559d1eef1133e2814de2ed56de49ac707af81b
Parents: 6c8f5d1
Author: cstella <cestella@gmail.com>
Authored: Tue Jun 21 10:40:06 2016 -0400
Committer: cstella <cestella@gmail.com>
Committed: Tue Jun 21 10:40:06 2016 -0400

----------------------------------------------------------------------
 .../apache/metron/common/writer/NoopWriter.java | 143 +++++++++++++++++++
 .../metron/common/writer/NoopWriterTest.java    |  48 +++++++
 2 files changed, 191 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/72559d1e/metron-platform/metron-common/src/main/java/org/apache/metron/common/writer/NoopWriter.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/main/java/org/apache/metron/common/writer/NoopWriter.java
b/metron-platform/metron-common/src/main/java/org/apache/metron/common/writer/NoopWriter.java
new file mode 100644
index 0000000..3b27f75
--- /dev/null
+++ b/metron-platform/metron-common/src/main/java/org/apache/metron/common/writer/NoopWriter.java
@@ -0,0 +1,143 @@
+/**
+ * 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.metron.common.writer;
+
+import backtype.storm.tuple.Tuple;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Iterables;
+import org.apache.metron.common.configuration.writer.WriterConfiguration;
+import org.apache.metron.common.interfaces.BulkMessageWriter;
+import org.apache.metron.common.utils.ConversionUtils;
+
+import java.io.Closeable;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.function.Function;
+
+public class NoopWriter extends AbstractWriter implements BulkMessageWriter<Object>
{
+
+  public static class RandomLatency implements Function<Void, Void> {
+    private int min;
+    private int max;
+
+    public RandomLatency(int min, int max) {
+      this.min = min;
+      this.max = max;
+    }
+
+    public int getMin() {
+      return min;
+    }
+    public int getMax() {
+      return max;
+    }
+
+    @Override
+    public Void apply(Void aVoid) {
+      int sleepMs = ThreadLocalRandom.current().nextInt(min, max + 1);
+      try {
+        Thread.sleep(sleepMs);
+      } catch (InterruptedException e) {
+      }
+      return null;
+    }
+  }
+
+  public static class FixedLatency implements Function<Void, Void> {
+    private int latency;
+    public FixedLatency(int latency) {
+      this.latency = latency;
+    }
+    public int getLatency() {
+      return latency;
+    }
+
+    @Override
+    public Void apply(Void aVoid) {
+      if(latency > 0) {
+        try {
+          Thread.sleep(latency);
+        } catch (InterruptedException e) {
+        }
+      }
+      return null;
+    }
+  }
+  Function<Void, Void> sleepFunction = null;
+
+  public NoopWriter withLatency(String sleepConfig) {
+    sleepFunction = getSleepFunction(sleepConfig);
+    return this;
+  }
+
+
+  private Function<Void, Void> getSleepFunction(String sleepConfig) {
+    String usageMessage = "Unexpected: " + sleepConfig + " Expected value: integer for a
fixed sleep duration in milliseconds (e.g. 10) " +
+            "or a range of latencies separated by a comma (e.g. \"10, 20\") to sleep a random
amount in that range.";
+    try {
+      if (sleepConfig.contains(",")) {
+        // random latency within a range.
+        Iterable<String> it = Splitter.on(',').split(sleepConfig);
+        Integer min = ConversionUtils.convert(Iterables.getFirst(it, "").trim(), Integer.class);
+        Integer max= ConversionUtils.convert(Iterables.getLast(it, "").trim(), Integer.class);
+        if (min != null && max != null) {
+          return new RandomLatency(min, max);
+        }
+      } else {
+        //fixed latency
+        Integer latency = ConversionUtils.convert(sleepConfig.trim(), Integer.class);
+        if(latency != null) {
+          return new FixedLatency(latency);
+        }
+      }
+    }
+    catch(Throwable t) {
+      throw new IllegalArgumentException(usageMessage, t);
+    }
+    throw new IllegalArgumentException(usageMessage);
+  }
+
+  @Override
+  public void configure(String sensorName, WriterConfiguration configuration) {
+    Map<String, Object> config = configuration.getSensorConfig(sensorName);
+    if(config != null) {
+      Object noopLatency = config.get("noopLatency");
+      if(noopLatency != null) {
+        sleepFunction = getSleepFunction(noopLatency.toString());
+      }
+    }
+  }
+
+  @Override
+  public void init(Map stormConf, WriterConfiguration config) throws Exception {
+  }
+
+  @Override
+  public void write(String sensorType, WriterConfiguration configurations, Iterable<Tuple>
tuples, List<Object> messages) throws Exception {
+    if(sleepFunction != null) {
+      sleepFunction.apply(null);
+    }
+  }
+
+  @Override
+  public void close() throws Exception {
+
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-metron/blob/72559d1e/metron-platform/metron-common/src/test/java/org/apache/metron/common/writer/NoopWriterTest.java
----------------------------------------------------------------------
diff --git a/metron-platform/metron-common/src/test/java/org/apache/metron/common/writer/NoopWriterTest.java
b/metron-platform/metron-common/src/test/java/org/apache/metron/common/writer/NoopWriterTest.java
new file mode 100644
index 0000000..e40ce7c
--- /dev/null
+++ b/metron-platform/metron-common/src/test/java/org/apache/metron/common/writer/NoopWriterTest.java
@@ -0,0 +1,48 @@
+/**
+ * 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.metron.common.writer;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class NoopWriterTest {
+  @Test
+  public void testFixedLatencyConfig() {
+    NoopWriter writer = new NoopWriter().withLatency("10");
+    Assert.assertTrue(writer.sleepFunction instanceof NoopWriter.FixedLatency);
+    NoopWriter.FixedLatency sleepFunction = (NoopWriter.FixedLatency)writer.sleepFunction;
+    Assert.assertEquals(10, sleepFunction.getLatency());
+  }
+
+  private void ensureRandomLatencyConfig(String latencyConfig, int min, int max) {
+    NoopWriter writer = new NoopWriter().withLatency(latencyConfig);
+    Assert.assertTrue(writer.sleepFunction instanceof NoopWriter.RandomLatency);
+    NoopWriter.RandomLatency sleepFunction = (NoopWriter.RandomLatency)writer.sleepFunction;
+    Assert.assertEquals(min, sleepFunction.getMin());
+    Assert.assertEquals(max, sleepFunction.getMax());
+  }
+
+  @Test
+  public void testRandomLatencyConfig() {
+    ensureRandomLatencyConfig("10,20", 10, 20);
+    ensureRandomLatencyConfig("10, 20", 10, 20);
+    ensureRandomLatencyConfig("10 ,20", 10, 20);
+    ensureRandomLatencyConfig("10 , 20", 10, 20);
+  }
+
+}


Mime
View raw message