import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
private void setLastCheckTime(final Long timeStamp) {
    Map<String, Object> statMap = Maps.newHashMap();
    statMap.put("lastChecked", timeStamp);
    client.prepareIndex(riverIndexName, riverName.getName(), "_stat").setSource(statMap).execute(new ActionListener<IndexResponse>() {
        @Override
        public void onResponse(IndexResponse indexResponse) {
            if (logger.isDebugEnabled()) {
                logger.debug("Last check time saved, {}", timeStamp);
            }
        }

        @Override
        public void onFailure(Throwable e) {
            logger.error("Failed to record last checked time stamp", e);
        }
    });
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Test
public void shouldGenerateObjectID() throws Exception {
	ExtendedElasticSearchLogStashEventSerializer fixture = new ExtendedElasticSearchLogStashEventSerializer();
	Map<String, String> parameters = new HashMap<String, String>();
	parameters.put("generateId", "true");
	Context context = new Context(parameters);
	fixture.configure(context);

	String message = "test body";
	Map<String, String> headers = Maps.newHashMap();
	long timestamp = System.currentTimeMillis();
	headers.put("timestamp", String.valueOf(timestamp));
	headers.put("source", "flume_tail_src");
	headers.put("host", "[email protected]");
	headers.put("src_path", "/tmp/test");
	headers.put("headerNameOne", "headerValueOne");
	headers.put("headerNameTwo", "headerValueTwo");
	headers.put("type", "sometype");
	Event event = EventBuilder.withBody(message.getBytes(charset));
	event.setHeaders(headers);
	XContentBuilder actual = fixture.getXContentBuilder(event);
	DocumentIdBuilder idBuilder = (DocumentIdBuilder)fixture;
	String id = idBuilder.getDocumentId(actual.bytes());
	assertNotNull(id);
			
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Before
 public void setUp() throws IOException {
   initMocks(this);
   BytesReference bytesReference = mock(BytesReference.class);
   BytesStream bytesStream = mock(BytesStream.class);

   when(nameBuilder.getIndexName(any(Event.class))).thenReturn(INDEX_NAME);
   when(bytesReference.toBytesArray()).thenReturn(new BytesArray(MESSAGE_CONTENT));
   when(bytesStream.bytes()).thenReturn(bytesReference);
   serializer = new ExtendedElasticSearchLogStashEventSerializer();
Map<String, String> headers = Maps.newHashMap();
String message = "body";
event = EventBuilder.withBody(message.getBytes(charset));
event.setHeaders(headers);
   fixture = new ElasticSearchRestClient(HOSTS, serializer, httpClient);
 }
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
private static Map<String, Object> getIndexableIdMap(Field field) {
    Map<String, Object> idMap = Maps.newHashMap();

    IndexableId indexableId = field.getAnnotation(IndexableId.class);
    if (indexableId.index() != IndexEnum.NA) {
        idMap.put("index", indexableId.index().toString().toLowerCase());
    }

    if (indexableId.store()) {
        idMap.put("store", "yes");
    }

    IndexableProperty indexableProperty = field.getAnnotation(IndexableProperty.class);
    if (indexableProperty != null) {
        String fieldName = field.getName();
        if (indexableProperty.name() != null && !indexableProperty.name().isEmpty()) {
            fieldName = indexableProperty.name();
        }
        idMap.put("path", fieldName);  // only need to put this if the IndexableId field is also IndexableProperty
    }

    return idMap;
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
protected void appendHeaders(XContentBuilder builder, Event event)
    throws IOException {
  Map<String, String> headers = Maps.newHashMap(event.getHeaders());

  String timestamp = headers.get("timestamp");
  if (!StringUtils.isBlank(timestamp)
      && StringUtils.isBlank(headers.get("timestamp"))) {
    long timestampMs = Long.parseLong(timestamp);
    builder.field("timestamp", new Date(timestampMs));
  }

  for(Map.Entry<String,String> entry: headers.entrySet()) {
    String value = entry.getValue();
    if(StringUtils.isNotBlank(value))
      builder.field(entry.getKey(), value);
  }

  builder.endObject();
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
public static ResultWithGroupings readFrom(StreamInput in) throws IOException {
    GeoHashCell cell = GeoHashCell.readFrom(in);
    long total = in.readLong();
    int groupingsSize = in.readInt();

    Map<String, AtomicLong> groupings = Maps.newHashMap();

    for (int i = 0; i<groupingsSize; i++) {
        String key = in.readString();
        long value = in.readLong();

        groupings.put(key, new AtomicLong(value));
    }

    return new GeoHashCellWithGroupings(cell, total, groupings);
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Override
public void parseAnnotation(ObjectField annotation, Map<String, Object> fieldDefinition, String pathPrefix, String nestedPrefix, Indexable indexable) {
    if (fieldDefinition.get("type") != null) {
        LOGGER.info("Overriding mapping for field {} for class {} was defined as type {}", indexable.getName(), indexable.getDeclaringClassName(),
                fieldDefinition.get("type"));
        fieldDefinition.clear();
    }

    Class<?> objectClass = annotation == null ? ObjectField.class : annotation.objectClass();
    Boolean enabled = annotation == null ? true : annotation.enabled();

    fieldDefinition.put("type", "object");
    fieldDefinition.put("enabled", enabled);
    if (enabled) {
        Map<String, SourceFetchContext> fetchContext = Maps.newHashMap();
        // nested types can provide replacement class to be managed. This can be usefull to override map default type for example.
        Class<?> replaceClass = objectClass.equals(ObjectField.class) ? indexable.getType() : objectClass;
        try {
            String newPrefix = pathPrefix == null ? indexable.getName() + "." : pathPrefix + indexable.getName() + ".";
            this.fieldsMappingBuilder.parseFieldMappings(replaceClass, fieldDefinition, facets, filters, fetchContext, newPrefix, nestedPrefix);
        } catch (IntrospectionException e) {
            LOGGER.error("Fail to parse object class <" + replaceClass.getName() + ">", e);
        }
    }
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Override
public void parseAnnotation(NestedObject annotation, Map<String, Object> fieldDefinition, String pathPrefix, String nestedPrefix, Indexable indexable) {
    if (fieldDefinition.get("type") != null) {
        LOGGER.info("Overriding mapping for field {} for class {} was defined as type {}", indexable.getName(), indexable.getDeclaringClassName(),
                fieldDefinition.get("type"));
        fieldDefinition.clear();
    }

    fieldDefinition.put("type", "nested");
    Map<String, SourceFetchContext> fetchContext = Maps.newHashMap();
    // nested types can provide replacement class to be managed. This can be usefull to override map default type for example.
    Class<?> replaceClass = annotation.nestedClass().equals(NestedObject.class) ? indexable.getType() : annotation.nestedClass();
    try {
        this.fieldsMappingBuilder.parseFieldMappings(replaceClass, fieldDefinition, facets, filters, fetchContext, indexable.getName() + ".",
                indexable.getName());
    } catch (IntrospectionException e) {
        LOGGER.error("Fail to parse nested class <" + replaceClass.getName() + ">", e);
    }
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
private String buildSettings(ObjectMapper mapper, IndexAnalyserDefinition[] customAnalyserDefinitions) throws JsonProcessingException {
    if (ArrayUtils.isEmpty(customAnalyserDefinitions)) {
        return null;
    }

    Map<Object, Object> analysers = Maps.newHashMap();
    for (IndexAnalyserDefinition analyserDefinition : customAnalyserDefinitions) {
        AnalyserFields analyserFields = new AnalyserFields();
        analyserFields.char_filter = analyserDefinition.char_filter();
        analyserFields.filter = analyserDefinition.filters();
        analyserFields.stopwords = analyserDefinition.stopwords();
        analyserFields.tokenizer = analyserDefinition.tokenizer();
        analyserFields.type = analyserDefinition.type();
        analysers.put(analyserDefinition.name(), analyserFields);
    }

    return "{\"analysis\":{\"analyzer\":" + mapper.writeValueAsString(analysers) + "}}";
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
public Map<String, Set<String>> tokenize(String content) {
    Map<String, Set<String>> namedEntities = Maps.newHashMap();

    List<TextAnnotation> allTextAnnotations = new ArrayList<TextAnnotation>();
    String[] tokens = SimpleTokenizer.INSTANCE.tokenize(content);
    for (Map.Entry<String, TokenNameFinderModel> finderEntry : finders.entrySet()) {
        String type = finderEntry.getKey();
        NameFinderME finder = new NameFinderME(finderEntry.getValue());

        Span[] spans = finder.find(tokens);
        double[] probs = finder.probs(spans);

        for (int ni = 0; ni < spans.length; ni++) {
            allTextAnnotations.add(new TextAnnotation(type, spans[ni], probs[ni]));
        }
    }

    if (allTextAnnotations.size() > 0 ) {
        removeConflicts(allTextAnnotations);
    }
    convertTextAnnotationsToNamedEntities(tokens, allTextAnnotations, namedEntities);

    return namedEntities;
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Before
public void setupMapperParser() {
    Index index = new Index("test");

    Map<String, AnalyzerProviderFactory> analyzerFactoryFactories = Maps.newHashMap();
    analyzerFactoryFactories.put("keyword", new PreBuiltAnalyzerProviderFactory("keyword", AnalyzerScope.INDEX, new KeywordAnalyzer()));
    AnalysisService analysisService = new AnalysisService(index, ImmutableSettings.Builder.EMPTY_SETTINGS, null, analyzerFactoryFactories, null, null, null);

    mapperParser = new DocumentMapperParser(index, analysisService, new PostingsFormatService(index),
            new SimilarityLookupService(index, ImmutableSettings.Builder.EMPTY_SETTINGS));
    Settings settings = settingsBuilder()
            .put("opennlp.models.name.file", "src/test/resources/models/en-ner-person.bin")
            .put("opennlp.models.date.file", "src/test/resources/models/en-ner-date.bin")
            .put("opennlp.models.location.file", "src/test/resources/models/en-ner-location.bin")
            .build();

    LogConfigurator.configure(settings);

    OpenNlpService openNlpService = new OpenNlpService(settings);
    openNlpService.start();
    mapperParser.putTypeParser(OpenNlpMapper.CONTENT_TYPE, new OpenNlpMapper.TypeParser(analysisService, openNlpService));
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private Map<String, Object> getFieldMap(String key, Map<String, Object> fields, boolean createNew) {
	Map<String, Object> fieldMap = null;
	Object field = fields.get(key);
	if (null == field) {
		if (createNew) {
			fieldMap = Maps.newHashMap();
			fields.put(key, fieldMap);
		}
	} else if (field instanceof Map) {
		fieldMap = (Map<String, Object>) field;
	}
	return fieldMap;
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Test
public void testRoundTrip() throws Exception {
	ExtendedElasticSearchLogStashEventSerializer fixture = new ExtendedElasticSearchLogStashEventSerializer();
	Context context = new Context();
	fixture.configure(context);

	String message = "test body";
	Map<String, String> headers = Maps.newHashMap();
	long timestamp = System.currentTimeMillis();
	headers.put("timestamp", String.valueOf(timestamp));
	headers.put("source", "flume_tail_src");
	headers.put("host", "[email protected]");
	headers.put("src_path", "/tmp/test");
	headers.put("headerNameOne", "headerValueOne");
	headers.put("headerNameTwo", "headerValueTwo");
	headers.put("type", "sometype");
	Event event = EventBuilder.withBody(message.getBytes(charset));
	event.setHeaders(headers);

	XContentBuilder expected = jsonBuilder().startObject();
	expected.field("@message", new String(message.getBytes(), charset));
	expected.field("@timestamp", new Date(timestamp));
	expected.field("@source", "flume_tail_src");
	expected.field("@type", "sometype");
	expected.field("@source_host", "[email protected]");
	expected.field("@source_path", "/tmp/test");
	expected.startObject("@fields");
	expected.field("headerNameTwo", "headerValueTwo");
	expected.field("headerNameOne", "headerValueOne");
	expected.endObject();

	expected.endObject();

	XContentBuilder actual = fixture.getXContentBuilder(event);
	assertEquals(new String(expected.bytes().array()), new String(actual.bytes().array()));
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Test
public void shouldRemoveFieldsPrefix() throws Exception {
	ExtendedElasticSearchLogStashEventSerializer fixture = new ExtendedElasticSearchLogStashEventSerializer();
	Map<String, String> parameters = new HashMap<String, String>();
	parameters.put("objectFields", "params, anotherField");
	parameters.put("removeFieldsPrefix", "true");
	Context context = new Context(parameters);
	fixture.configure(context);

	String message = "test body";
	Map<String, String> headers = Maps.newHashMap();
	long timestamp = System.currentTimeMillis();
	headers.put("timestamp", String.valueOf(timestamp));
	headers.put("source", "flume_tail_src");
	headers.put("host", "[email protected]");
	headers.put("src_path", "/tmp/test");
	headers.put("headerNameOne", "headerValueOne");
	headers.put("headerNameTwo", "headerValueTwo");
	headers.put("type", "sometype");
	Event event = EventBuilder.withBody(message.getBytes(charset));
	event.setHeaders(headers);

	XContentBuilder expected = jsonBuilder().startObject();
	expected.field("@message", new String(message.getBytes(), charset));
	expected.field("@timestamp", new Date(timestamp));
	expected.field("@source", "flume_tail_src");
	expected.field("@type", "sometype");
	expected.field("@source_host", "[email protected]");
	expected.field("@source_path", "/tmp/test");
	expected.field("headerNameTwo", "headerValueTwo");
	expected.field("headerNameOne", "headerValueOne");

	expected.endObject();

	XContentBuilder actual = fixture.getXContentBuilder(event);
	assertEquals(new String(expected.bytes().array()), new String(actual.bytes().array()));
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
public void shouldHandleInvalidJSONDuringComplexParsing() throws Exception {
	ExtendedElasticSearchLogStashEventSerializer fixture = new ExtendedElasticSearchLogStashEventSerializer();
	Context context = new Context();
	fixture.configure(context);

	String message = "{flume: somethingnotvalid}";
	Map<String, String> headers = Maps.newHashMap();
	long timestamp = System.currentTimeMillis();
	headers.put("timestamp", String.valueOf(timestamp));
	headers.put("source", "flume_tail_src");
	headers.put("host", "[email protected]");
	headers.put("src_path", "/tmp/test");
	headers.put("headerNameOne", "headerValueOne");
	headers.put("headerNameTwo", "headerValueTwo");
	headers.put("type", "sometype");
	Event event = EventBuilder.withBody(message.getBytes(charset));
	event.setHeaders(headers);

	XContentBuilder expected = jsonBuilder().startObject();
	expected.field("@message", new String(message.getBytes(), charset));
	expected.field("@timestamp", new Date(timestamp));
	expected.field("@source", "flume_tail_src");
	expected.field("@type", "sometype");
	expected.field("@source_host", "[email protected]");
	expected.field("@source_path", "/tmp/test");
	expected.startObject("@fields");
	expected.field("headerNameTwo", "headerValueTwo");
	expected.field("headerNameOne", "headerValueOne");
	expected.endObject();
	expected.endObject();

	XContentBuilder actual = fixture.getXContentBuilder(event);
	String expectedStr = new String(expected.bytes().array());
	String actualStr = new String(actual.bytes().array());
	assertEquals(expectedStr, actualStr);
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Test
public void shouldHandleSONInHeaderAsRaw() throws Exception {
	ExtendedElasticSearchLogStashEventSerializer fixture = new ExtendedElasticSearchLogStashEventSerializer();
	Map<String, String> parameters = new HashMap<String, String>();
	Context context = new Context(parameters);
	fixture.configure(context);

	String message = "{flume: somethingnotvalid}";
	String params = "{\"cmd\":\"api.method\",\"email\":\"[email protected]\"}";
	Map<String, String> headers = Maps.newHashMap();
	long timestamp = System.currentTimeMillis();
	headers.put("timestamp", String.valueOf(timestamp));
	headers.put("params", params);
	Event event = EventBuilder.withBody(message.getBytes(charset));
	event.setHeaders(headers);

	XContentBuilder expected = jsonBuilder().startObject();
	expected.field("@message", new String(message.getBytes(), charset));
	expected.field("@timestamp", new Date(timestamp));
	expected.startObject("@fields");
	expected.field("params", params);
	expected.endObject();
	expected.endObject();

	XContentBuilder actual = fixture.getXContentBuilder(event);
	String expectedStr = new String(expected.bytes().array());
	String actualStr = new String(actual.bytes().array());
	assertEquals(expectedStr, actualStr);
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Test
public void shouldParseJSONInHeader() throws Exception {
	ExtendedElasticSearchLogStashEventSerializer fixture = new ExtendedElasticSearchLogStashEventSerializer();
	Map<String, String> parameters = new HashMap<String, String>();
	parameters.put("objectFields", "params, anotherField");
	Context context = new Context(parameters);
	fixture.configure(context);

	String message = "{flume: somethingnotvalid}";
	String params = "{\"cmd\":\"api.method\",\"email\":\"[email protected]\"}";
	Map<String, String> headers = Maps.newHashMap();
	long timestamp = System.currentTimeMillis();
	headers.put("timestamp", String.valueOf(timestamp));
	headers.put("params", params);
	Event event = EventBuilder.withBody(message.getBytes(charset));
	event.setHeaders(headers);

	XContentBuilder expected = jsonBuilder().startObject();
	expected.field("@message", new String(message.getBytes(), charset));
	expected.field("@timestamp", new Date(timestamp));
	expected.startObject("@fields");
	expected.startObject("params");
	expected.field("cmd", "api.method");
	expected.field("email", "[email protected]");
	expected.endObject();
	expected.endObject();
	expected.endObject();

	XContentBuilder actual = fixture.getXContentBuilder(event);
	String expectedStr = new String(expected.bytes().array());
	String actualStr = new String(actual.bytes().array());
	assertEquals(expectedStr, actualStr);
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Test
public void shouldParseNestedJSONInHeader() throws Exception {
	ExtendedElasticSearchLogStashEventSerializer fixture = new ExtendedElasticSearchLogStashEventSerializer();
	Map<String, String> parameters = new HashMap<String, String>();
	parameters.put("objectFields", "params, anotherField");
	Context context = new Context(parameters);
	fixture.configure(context);

	String message = "{flume: somethingnotvalid}";
	String params = "{\"cmd\":\"api.method\",\"email\":\"[email protected]\", \"nested\" : { \"sub\" : 1 }}";
	Map<String, String> headers = Maps.newHashMap();
	long timestamp = System.currentTimeMillis();
	headers.put("timestamp", String.valueOf(timestamp));
	headers.put("params", params);
	Event event = EventBuilder.withBody(message.getBytes(charset));
	event.setHeaders(headers);

	XContentBuilder expected = jsonBuilder().startObject();
	expected.field("@message", new String(message.getBytes(), charset));
	expected.field("@timestamp", new Date(timestamp));
	expected.startObject("@fields");
	expected.startObject("params");
	expected.field("cmd", "api.method");
	expected.startObject("nested");
	expected.field("sub", 1);
	expected.endObject();
	expected.field("email", "[email protected]");
	expected.endObject();
	expected.endObject();
	expected.endObject();

	XContentBuilder actual = fixture.getXContentBuilder(event);
	String expectedStr = new String(expected.bytes().array());
	String actualStr = new String(actual.bytes().array());
	assertEquals(expectedStr, actualStr);
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Test
public void shouldParseNestedJSONInHeaderWhenCollating() throws Exception {
	ExtendedElasticSearchLogStashEventSerializer fixture = new ExtendedElasticSearchLogStashEventSerializer();
	Map<String, String> parameters = new HashMap<String, String>();
	parameters.put("objectFields", "params");
	parameters.put("collateObjects", "true");
	Context context = new Context(parameters);
	fixture.configure(context);

	String message = "{flume: somethingnotvalid}";
	String params = "{\"cmd\":\"api.method\",\"email\":\"[email protected]\", \"nested\" : { \"sub\" : 1 }}";
	Map<String, String> headers = Maps.newHashMap();
	long timestamp = System.currentTimeMillis();
	headers.put("timestamp", String.valueOf(timestamp));
	headers.put("params", params);
	Event event = EventBuilder.withBody(message.getBytes(charset));
	event.setHeaders(headers);

	XContentBuilder expected = jsonBuilder().startObject();
	expected.field("@message", new String(message.getBytes(), charset));
	expected.field("@timestamp", new Date(timestamp));
	expected.startObject("@fields");
	expected.startObject("params");
	expected.field("cmd", "api.method");
	expected.startObject("nested");
	expected.field("sub", 1);
	expected.endObject();
	expected.field("email", "[email protected]");
	expected.endObject();
	expected.endObject();
	expected.endObject();

	XContentBuilder actual = fixture.getXContentBuilder(event);
	String expectedStr = new String(expected.bytes().array());
	String actualStr = new String(actual.bytes().array());
	assertEquals(expectedStr, actualStr);
}
 

import org.elasticsearch.common.collect.Maps; //导入依赖的package包/类
@Test
public void shouldCollateJSONDefaultOneLevel() throws Exception {
	ExtendedElasticSearchLogStashEventSerializer fixture = new ExtendedElasticSearchLogStashEventSerializer();
	Map<String, String> parameters = new HashMap<String, String>();
	parameters.put("objectFields", "params, anotherField");
	parameters.put("collateObjects", "true");
	Context context = new Context(parameters);
	fixture.configure(context);

	String message = "{flume: somethingnotvalid}";
	Map<String, String> headers = Maps.newHashMap();
	long timestamp = System.currentTimeMillis();
	headers.put("timestamp", String.valueOf(timestamp));
	headers.put("params.cmd", "api.call");
	headers.put("params.email", "[email protected]");
	headers.put("params.timer.start", "1");
	headers.put("params.timer.end", "2");
	Event event = EventBuilder.withBody(message.getBytes(charset));
	event.setHeaders(headers);

	XContentBuilder expected = jsonBuilder().startObject();
	expected.field("@message", new String(message.getBytes(), charset));
	expected.field("@timestamp", new Date(timestamp));
	expected.startObject("@fields");
	expected.startObject("params");
	expected.field("timer.start", "1");
	expected.field("cmd", "api.call");
	expected.field("email", "[email protected]");
	expected.field("timer.end", "2");
	expected.endObject();
	expected.endObject();
	expected.endObject();

	XContentBuilder actual = fixture.getXContentBuilder(event);
	String expectedStr = new String(expected.bytes().array());
	String actualStr = new String(actual.bytes().array());
	assertEquals(expectedStr, actualStr);
}