Merge pull request #242 from jiafu1115/patch-23

support writing Point by async/sync through UDP.

Author: majst01

README.md

@@ -74,8 +74,19 @@ influxDB.deleteDatabase(dbName);
 ```
 Note that the batching functionality creates an internal thread pool that needs to be shutdown explicitly as part of a gracefull application shut-down, or the application will not shut down properly. To do so simply call: ```influxDB.close()```
 
+Gzip's support:
+
 influxdb-java client doesn't enable gzip compress for http request body by default. If you want to enable gzip to reduce transfter data's size , you can call: ```influxDB.enableGzip()```
 
+UDP's support:
+
+influxdb-java client support udp protocol now. you can call followed methods directly to wirte through UDP.
+```
+public void write(final int udpPort, final String records);
+public void write(final int udpPort, final List<String> records);
+public void write(final int udpPort, final Point point);
+```
+
 ### Changes in 2.4
 influxdb-java now uses okhttp3 and retrofit2.  As a result, you can now pass an ``OkHttpClient.Builder``
 to the ``InfluxDBFactory.connect`` if you wish to add more interceptors, etc, to OkHttp.

src/main/java/org/influxdb/InfluxDB.java

@@ -153,6 +153,16 @@ public InfluxDB enableBatch(final int actions, final int flushDuration, final Ti
    */
   public void write(final String database, final String retentionPolicy, final Point point);
 
+  /**
+   * Write a single Point to the database through UDP.
+   *
+   * @param udpPort
+   *            the udpPort to write to.
+   * @param point
+   *            The point to write.
+   */
+  public void write(final int udpPort, final Point point);
+
   /**
    * Write a set of Points to the influxdb database with the new (>= 0.9.0rc32) lineprotocol.
    *

src/main/java/org/influxdb/impl/BatchProcessor.java

@@ -3,6 +3,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -187,10 +188,12 @@ void write() {
       }
 
       Map<String, BatchPoints> databaseToBatchPoints = Maps.newHashMap();
+      Map<Integer, List<String>> udpPortToBatchPoints = Maps.newHashMap();
       List<AbstractBatchEntry> batchEntries = new ArrayList<>(this.queue.size());
       this.queue.drainTo(batchEntries);
 
       for (AbstractBatchEntry batchEntry : batchEntries) {
+        Point point = batchEntry.getPoint();
         if (batchEntry instanceof HttpBatchEntry) {
             HttpBatchEntry httpBatchEntry = HttpBatchEntry.class.cast(batchEntry);
             String dbName = httpBatchEntry.getDb();
@@ -199,15 +202,24 @@ void write() {
                                                    .retentionPolicy(httpBatchEntry.getRp()).build();
               databaseToBatchPoints.put(dbName, batchPoints);
             }
-            Point point = batchEntry.getPoint();
             databaseToBatchPoints.get(dbName).point(point);
+        } else if (batchEntry instanceof UdpBatchEntry) {
+            UdpBatchEntry udpBatchEntry = UdpBatchEntry.class.cast(batchEntry);
+            int udpPort = udpBatchEntry.getUdpPort();
+            if (!udpPortToBatchPoints.containsKey(udpPort)) {
+              List<String> batchPoints = new ArrayList<String>();
+              udpPortToBatchPoints.put(udpPort, batchPoints);
+            }
+            udpPortToBatchPoints.get(udpPort).add(point.lineProtocol());
         }
-
       }
 
       for (BatchPoints batchPoints : databaseToBatchPoints.values()) {
         BatchProcessor.this.influxDB.write(batchPoints);
       }
+      for (Entry<Integer, List<String>> entry : udpPortToBatchPoints.entrySet()) {
+          BatchProcessor.this.influxDB.write(entry.getKey(), entry.getValue());
+      }
     } catch (Throwable t) {
       // any exception wouldn't stop the scheduler
       LOG.log(Level.SEVERE, "Batch could not be sent. Data will be lost", t);

src/main/java/org/influxdb/impl/InfluxDBImpl.java

@@ -13,6 +13,8 @@
 import org.influxdb.dto.Query;
 import org.influxdb.dto.QueryResult;
 import org.influxdb.impl.BatchProcessor.HttpBatchEntry;
+import org.influxdb.impl.BatchProcessor.UdpBatchEntry;
+
 import okhttp3.Headers;
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
@@ -220,6 +222,21 @@ public void write(final String database, final String retentionPolicy, final Poi
     this.writeCount.incrementAndGet();
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public void write(final int udpPort, final Point point) {
+    if (this.batchEnabled.get()) {
+      UdpBatchEntry batchEntry = new UdpBatchEntry(point, udpPort);
+      this.batchProcessor.put(batchEntry);
+    } else {
+      this.write(udpPort, point.lineProtocol());
+      this.unBatchedCount.incrementAndGet();
+    }
+    this.writeCount.incrementAndGet();
+  }
+
   @Override
   public void write(final BatchPoints batchPoints) {
     this.batchedCount.addAndGet(batchPoints.getPoints().size());

src/test/java/org/influxdb/InfluxDBTest.java

@@ -147,6 +147,41 @@ public void testWrite() {
 		Assert.assertFalse(result.getResults().get(0).getSeries().get(0).getTags().isEmpty());
 		this.influxDB.deleteDatabase(dbName);
 	}
+	
+	/**
+	 *  Test the implementation of {@link InfluxDB#write(int, Point)}'s sync support.
+	 */
+	@Test
+	public void testSyncWritePointThroughUDP() {
+		this.influxDB.disableBatch();
+		String measurement = TestUtils.getRandomMeasurement();
+		Point point = Point.measurement(measurement).tag("atag", "test").addField("used", 80L).addField("free", 1L).build();
+		this.influxDB.write(UDP_PORT, point);
+		Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
+		Query query = new Query("SELECT * FROM " + measurement + " GROUP BY *", UDP_DATABASE);
+		QueryResult result = this.influxDB.query(query);
+		Assert.assertFalse(result.getResults().get(0).getSeries().get(0).getTags().isEmpty());
+	}
+	
+	/**
+	 *  Test the implementation of {@link InfluxDB#write(int, Point)}'s async support.
+	 */
+	@Test
+	public void testAsyncWritePointThroughUDP() {
+		this.influxDB.enableBatch(1, 1, TimeUnit.SECONDS);
+		try{
+			Assert.assertTrue(this.influxDB.isBatchEnabled());
+			String measurement = TestUtils.getRandomMeasurement();
+			Point point = Point.measurement(measurement).tag("atag", "test").addField("used", 80L).addField("free", 1L).build();
+			this.influxDB.write(UDP_PORT, point);
+			Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
+			Query query = new Query("SELECT * FROM " + measurement + " GROUP BY *", UDP_DATABASE);
+			QueryResult result = this.influxDB.query(query);
+			Assert.assertFalse(result.getResults().get(0).getSeries().get(0).getTags().isEmpty());
+		}finally{
+			this.influxDB.disableBatch();
+		}
+	}
 
     /**
      * Test writing to the database using string protocol.