1 /***********************************************************************
2  * Copyright (c) 2013-2024 Commonwealth Computer Research, Inc.
3  * All rights reserved. This program and the accompanying materials
4  * are made available under the terms of the Apache License, Version 2.0
5  * which accompanies this distribution and is available at
6  * http://www.opensource.org/licenses/apache2.0.php.
7  ***********************************************************************/
8 
9 package org.locationtech.geomesa.fs.storage.common.utils
10 
11 import org.apache.commons.io.IOUtils
12 import org.geotools.api.feature.simple.SimpleFeatureType
13 import org.locationtech.geomesa.fs.storage.api.{NamedOptions, PartitionSchemeFactory}
14 import org.locationtech.geomesa.fs.storage.common.StorageSerialization
15 import org.locationtech.geomesa.fs.storage.common.utils.PartitionSchemeArgResolver.SchemeArgs
16 import org.locationtech.geomesa.utils.conf.ArgResolver
17 import org.locationtech.geomesa.utils.io.{PathUtils, WithClose}
18 
19 import java.nio.charset.StandardCharsets
20 import scala.util.control.NonFatal
21 
22 object PartitionSchemeArgResolver {
23 
24   private val instance = new PartitionSchemeArgResolver
25 
26   case class SchemeArgs(sft: SimpleFeatureType, scheme: String)
27 
28   def resolve(sft: SimpleFeatureType, scheme: String): Either[Throwable, NamedOptions] =
29     instance.getArg(SchemeArgs(sft, scheme))
30 }
31 
32 class PartitionSchemeArgResolver extends ArgResolver[NamedOptions, SchemeArgs] {
33 
34   import org.locationtech.geomesa.utils.conf.ArgResolver.ArgTypes._
35 
36   private val fileNameReg = """([^.]*)\.([^.]*)""" // e.g. "foo.bar"
37   private val confStrings = Seq("{")
38 
39   override protected def argType(args: SchemeArgs): ArgTypes = {
40     // guess the type we are trying to parse, to determine which error we show for failures
41     // order is important here
42     if (confStrings.exists(args.scheme.contains)) {
43       CONFSTR
44     } else if (args.scheme.matches(fileNameReg) || args.scheme.contains("/")) {
45       PATH
46     } else {
47       NAME
48     }
49   }
50 
51   override val parseMethodList: Seq[SchemeArgs => ResEither] = List[SchemeArgs => ResEither](
52     getNamedScheme,
53     parseFile,
54     parseString
55   )
56 
57   private [PartitionSchemeArgResolver] def getNamedScheme(args: SchemeArgs): ResEither = {
58     try {
59       val options = NamedOptions(args.scheme)
60       PartitionSchemeFactory.load(args.sft, options)
61       Right(options)
62     } catch {
63       case NonFatal(e) => Left((s"Unable to load named scheme ${args.scheme}", e, NAME))
64     }
65   }
66 
67   private [PartitionSchemeArgResolver] def parseString(args: SchemeArgs): ResEither = {
68     try {
69       val options = StorageSerialization.deserialize(args.scheme)
70       PartitionSchemeFactory.load(args.sft, options)
71       Right(options)
72     } catch {
73       case NonFatal(e) => Left((s"Unable to load scheme from arg ${args.scheme}", e, CONFSTR))
74     }
75   }
76 
77   private [PartitionSchemeArgResolver] def parseFile(args: SchemeArgs): ResEither = {
78     try {
79       val handle = PathUtils.interpretPath(args.scheme).headOption.getOrElse {
80         throw new RuntimeException(s"Could not read file at ${args.scheme}")
81       }
82       WithClose(handle.open) { is =>
83         if (is.hasNext) {
84           val config = IOUtils.toString(is.next._2, StandardCharsets.UTF_8)
85           val options = StorageSerialization.deserialize(config)
86           PartitionSchemeFactory.load(args.sft, options)
87           Right(options)
88         } else {
89           throw new RuntimeException(s"Could not read file at ${args.scheme}")
90         }
91       }
92     } catch {
93       case NonFatal(e) => Left((s"Unable to load scheme from file ${args.scheme}", e, PATH))
94     }
95   }
96 }