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.kafka.streams
10 
11 import org.locationtech.geomesa.kafka.streams.MessageAction.MessageAction
12 
13 /**
14  * Data model for a GeoMesa data store message, used as the value in a Kafka record
15  *
16  * @param action message action
17  * @param attributes attributes of the simple feature represented by this message
18  * @param userData user data of the simple feature represented by this message
19  */
20 case class GeoMesaMessage(action: MessageAction, attributes: Seq[AnyRef], userData: Map[String, String] = Map.empty) {
21 
22   import scala.collection.JavaConverters._
23 
24   def asJava(): java.util.List[AnyRef] = attributes.asJava
25 }
26 
27 object GeoMesaMessage {
28 
29   import scala.collection.JavaConverters._
30 
31   /**
32    * Create an upsert message
33    *
34    * @param attributes feature attribute values
35    * @return
36    */
37   def upsert(attributes: Seq[AnyRef]): GeoMesaMessage = GeoMesaMessage(MessageAction.Upsert, attributes)
38 
39   /**
40    * Create an upsert message
41    *
42    * @param attributes feature attribute values
43    * @return
44    */
45   def upsert(attributes: java.util.List[AnyRef]): GeoMesaMessage = upsert(attributes.asScala.toSeq)
46 
47   /**
48    * Create an upsert message
49    *
50    * @param attributes feature attribute values
51    * @param userData feature user data
52    * @return
53    */
54   def upsert(attributes: Seq[AnyRef], userData: Map[String, String]): GeoMesaMessage =
55     GeoMesaMessage(MessageAction.Upsert, attributes, userData)
56 
57   /**
58    * Create an upsert message
59    *
60    * @param attributes feature attribute values
61    * @return
62    */
63   def upsert(attributes: java.util.List[AnyRef], userData: java.util.Map[String, String]): GeoMesaMessage =
64     upsert(attributes.asScala.toSeq, userData.asScala.toMap)
65 
66   /**
67    * Create a delete message
68    *
69    * @return
70    */
71   def delete(): GeoMesaMessage = GeoMesaMessage(MessageAction.Delete, Seq.empty)
72 }