Add support for Azure metadata

Author: blootsvoets

src/main/java/org/radarbase/output/source/AzureSourceStorage.kt

@@ -9,7 +9,6 @@ import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.Paths
 
-
 class AzureSourceStorage(
         client: BlobServiceClient,
         container: String,
@@ -22,6 +21,28 @@ class AzureSourceStorage(
             .asSequence()
             .map { SimpleFileStatus(Paths.get(it.name), it.isPrefix ?: false, it.properties?.lastModified?.toInstant()) }
 
+
+    override fun createTopicFile(topic: String, status: SimpleFileStatus): TopicFile {
+        var topicFile = super.createTopicFile(topic, status)
+
+        if (topicFile.range.range.to == null) {
+            try {
+                val endOffset = blobClient(topicFile.path).properties.metadata["endOffset"]?.toLongOrNull()
+
+                if (endOffset != null) {
+                    topicFile = topicFile.copy(
+                            range = topicFile.range.mapRange {
+                                it.copy(to = endOffset)
+                            })
+                }
+            } catch (ex: Exception) {
+                // never mind
+            }
+        }
+
+        return topicFile
+    }
+
     override fun delete(path: Path) {
         blobClient(path).delete()
     }

src/main/java/org/radarbase/output/source/S3SourceStorage.kt

@@ -37,7 +37,10 @@ class S3SourceStorage(
             val tags = s3Client.getObjectTags(GetObjectTagsArgs.Builder().objectBuild(bucket, status.path))
             val endOffset = tags.get()["endOffset"]?.toLongOrNull()
             if (endOffset != null) {
-                topicFile = topicFile.copy(range = topicFile.range.mapRange { it.copy(to = endOffset) })
+                topicFile = topicFile.copy(
+                        range = topicFile.range.mapRange {
+                            it.copy(to = endOffset)
+                        })
             }
         }