mikayla-yang commented on a change in pull request #5476: URL: https://github.com/apache/nifi/pull/5476#discussion_r749780399
########## File path: nifi-nar-bundles/nifi-geohash-bundle/nifi-geohash-processors/src/main/java/org/apache/nifi/processors/geohash/GeohashRecord.java ########## @@ -0,0 +1,449 @@ +/* + * 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.nifi.processors.geohash; + +import org.apache.nifi.annotation.behavior.EventDriven; +import org.apache.nifi.annotation.behavior.SideEffectFree; +import org.apache.nifi.annotation.behavior.SupportsBatching; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.expression.ExpressionLanguageScope; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.flowfile.attributes.CoreAttributes; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.ProcessorInitializationContext; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.record.path.FieldValue; +import org.apache.nifi.record.path.RecordPath; +import org.apache.nifi.record.path.RecordPathResult; +import org.apache.nifi.record.path.util.RecordPathCache; +import org.apache.nifi.record.path.validation.RecordPathValidator; + +import org.apache.nifi.serialization.RecordReader; +import org.apache.nifi.serialization.RecordReaderFactory; +import org.apache.nifi.serialization.RecordSetWriter; +import org.apache.nifi.serialization.RecordSetWriterFactory; +import org.apache.nifi.serialization.record.Record; +import org.apache.nifi.serialization.WriteResult; + +import ch.hsr.geohash.GeoHash; +import ch.hsr.geohash.WGS84Point; + +import java.io.InputStream; +import java.io.OutputStream; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Set; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.Map; +import java.util.HashMap; +import java.util.Optional; + +@EventDriven +@SideEffectFree +@SupportsBatching +@Tags({"geo", "geohash", "record"}) +@CapabilityDescription("A record-based processor that encodes and decodes Geohashes from and to latitude/longitude coordinates.") +@InputRequirement(InputRequirement.Requirement.INPUT_REQUIRED) +public class GeohashRecord extends AbstractProcessor { + + public enum ProcessingMode { + ENCODE, DECODE + } + + public enum GeohashFormat { + BASE_32, BINARY_STRING, LONG + } + + public enum RoutingStrategy { + SKIP_UNENRICHED, + SPLIT, + REQUIRE_ALL_ENRICHED + } + + public static final PropertyDescriptor MODE = new PropertyDescriptor.Builder() + .name("mode") + .displayName("Mode") + .description("Specifies whether to encode latitude/longitude to geohash or decode geohash to latitude/longitude") + .required(true) + .allowableValues(ProcessingMode.values()) + .defaultValue(ProcessingMode.ENCODE.name()) + .build(); + + public static final PropertyDescriptor ROUTING_STRATEGY = new PropertyDescriptor.Builder() + .name("routing-strategy") + .displayName("Routing Strategy") + .description("Specifies how to route records after encoding or decoding has been performed. " + + "Skip_Unenriched will route a flowfile to success if any of its records is enriched, otherwise it will be sent to failure. " + + "Split_Enriched_Unenriched will separate the records that have been enriched from those have not and send them to success, while unenriched records will be sent to failure; " + + "and the original flowfile will be sent to the original relationship. " + + "Require_All_Enriched will route a flowfile to success only if all of its records are enriched, otherwise it will be sent to failure") + .required(true) + .allowableValues(RoutingStrategy.values()) + .defaultValue(RoutingStrategy.SKIP_UNENRICHED.name()) + .build(); + + public static final PropertyDescriptor RECORD_READER = new PropertyDescriptor.Builder() + .name("record-reader") + .displayName("Record Reader") + .description("Specifies the record reader service to use for reading incoming data") + .required(true) + .identifiesControllerService(RecordReaderFactory.class) + .build(); + + public static final PropertyDescriptor RECORD_WRITER = new PropertyDescriptor.Builder() + .name("record-writer") + .displayName("Record Writer") + .description("Specifies the record writer service to use for writing data") + .required(true) + .identifiesControllerService(RecordSetWriterFactory.class) + .build(); + + public static final PropertyDescriptor LATITUDE_RECORD_PATH = new PropertyDescriptor.Builder() + .name("latitude-record-path") + .displayName("Latitude Record Path") + .description("In the encode mode, this property specifies the record path to retrieve the latitude value; " + + "in the decode mode, this property specifies the record path to put the latitude value") + .required(true) + .addValidator(new RecordPathValidator()) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .build(); + + public static final PropertyDescriptor LONGITUDE_RECORD_PATH = new PropertyDescriptor.Builder() + .name("longitude-record-path") + .displayName("Longitude Record Path") + .description("In the encode mode, this property specifies the record path to retrieve the longitude value; " + + "in the decode mode, this property specifies the record path to put the longitude value") + .required(true) + .addValidator(new RecordPathValidator()) + .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES) + .build(); + + public static final PropertyDescriptor GEOHASH_FORMAT = new PropertyDescriptor.Builder() + .name("geohash-format") + .displayName("Geohash Format") + .description("In the encode mode, this property specifies the desired format for encoding geohash; " + + "in the decode mode, this property specifies the format of geohash provided") + .required(true) + .allowableValues(GeohashFormat.values()) + .defaultValue(GeohashFormat.BASE_32.name()) Review comment: GeohashFormat dose not support expression language because it uses allowableValues(), just like the property MODE and ROUTING_STRATEGY -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
