InfluxDBClient { public Func1<List<JsonEvent>, Observable<List<JsonEvent>>> client(Map map) { final MapWrap config = MapWrap.of(map).assertExists("fields", "db", "url", "user", "password"); final StringTemplate measurementTemplate = config.asStringTemplate("measurement"); final StringTemplate dbTemplate = config.asStringTemplate("db"); final InfluxDB influxDB = dbFactory.createOrGet(config); return events -> Observable.from(events) .groupBy(e -> dbTemplate.format(e).get()) .flatMap(byDatabase -> ensureDatabaseExists (influxDB, byDatabase)) .flatMap(byDatabase -> byDatabase .flatMap(jsonEvent -> buildPoint(config, measurementTemplate, jsonEvent)) .buffer(config.asInt("flushSize", 100)) .map(points -> toBatchPoints (byDatabase, points)) .flatMap(batchPoints -> save(batchPoints, influxDB)) ) .flatMap(o -> Observable.just(events)); } InfluxDBClient(); private  InfluxDBClient(Factory factory); static InfluxDBClient prepareForTest(InfluxDBClient.Factory factory); Func1<List<JsonEvent>, Observable<List<JsonEvent>>> client(Map map);  }
@Test public void test_save_metrics () { createClient (invocation -> { BatchPoints batchPoints = (BatchPoints) invocation.getArguments ()[0]; if (batchPoints.getDatabase ().equals ("thedatabase")) { List<Point> points = batchPoints.getPoints (); assertThat (points.size ()).isEqualTo (2); assertThat (points.get (0).lineProtocol ()).isEqualTo ("measurem,apiKey=thedatabase count=67.0 9223372036854775807"); } else if (batchPoints.getDatabase ().equals ("thedatabase2")) { List<Point> points = batchPoints.getPoints (); assertThat (points.size ()).isEqualTo (1); } return Void.TYPE; } ).client (new HashMap<String, Object> () {{ put ("url", "http: put ("measurement", "measurem"); put ("db", "{apiKey}"); put ("user", "root"); put ("password", "root"); put ("fields", MapWrap.of ("{metric}", "someValue").toMap ()); put ("flushSize", 100); put ("excludeTags", asList ("@timestamp", "message", "metric", "someValue", "time")); put ("time", "time"); put ("precision", TimeUnit.MILLISECONDS); }}) .call ( Lists.newArrayList ( Codecs.TEXT_TO_JSON.from ("hello") .put ("apiKey", "thedatabase") .put ("metric", "count") .put ("someValue", 67) .put ("time", 1473073541377000000L), Codecs.JSON_OBJECT.from ("{}") .put ("apiKey", "thedatabase") .put ("metric", "bytes") .put ("someValue", 67) .put ("@timestamp", "2015-08-30T15:28:11.561+02:00"), Codecs.JSON_OBJECT.from ("{}") .put ("apiKey", "thedatabase2") .put ("metric", "bytes") .put ("someValue", 67)) ); }