WIP on big data stores

Author: GearsAD

src/BigData.jl

@@ -0,0 +1,16 @@
+include("entities/AbstractDataStore.jl")
+include("entities/DataEntries.jl")
+include("entities/InMemoryDataStore.jl")
+include("entities/FileDataStore.jl")
+
+include("services/AbstractDataStore.jl")
+include("services/InMemoryDataStore.jl")
+include("services/FileDataStore.jl")
+
+export AbstractDataStore
+
+export GeneralBigDataEntry, MongodbBigDataEntry, FileBigDataEntry
+export InMemoryDataStore, FileDataStore
+
+export getBigData, addBigData!, updateBigData!, deleteBigData!, listStoreEntries
+export copyStore

src/BigData/BigData.jl

@@ -0,0 +1,7 @@
+
+"""
+    $(TYPEDEF)
+Superclass of all key-value datastores.
+"""
+abstract type AbstractDataStore{T}
+end

src/BigData/entities/AbstractDataStore.jl

@@ -0,0 +1,72 @@
+import Base: ==
+
+"""
+    $(TYPEDEF)
+GeneralBigDataEntry is a generic multipurpose data entry that creates a unique
+reproducible key using userId_robotId_sessionId_variableId_key.
+"""
+mutable struct GeneralBigDataEntry <: AbstractBigDataEntry
+    key::Symbol
+    originalKey::Symbol
+    createdTimestamp::DateTime
+    lastUpdatedTimestamp::DateTime
+    mimeType::String
+end
+
+## TODO: Move this
+function ==(a::GeneralBigDataEntry, b::GeneralBigDataEntry)
+    return a.key == b.key &&
+        a.originalKey == b.originalKey &&
+        a.mimeType == b.mimeType &&
+        Dates.value(a.createdTimestamp - b.createdTimestamp) < 1000 &&
+        Dates.value(a.lastUpdatedTimestamp - b.lastUpdatedTimestamp) < 1000 #1 second
+end
+
+
+# Clean up and move this.
+"""
+    $(SIGNATURES)
+Generates a unique key for the entry - userId_robotId_sessionId_variable_key.
+Simple.
+"""
+function uniqueKey(dfg::G, v::V, key::Symbol) where {G <: AbstractDFG, V <: AbstractDFGVariable}
+    key = join(String.([dfg.userId, dfg.robotId, dfg.sessionId, label(v), String(key)]), "_")
+    return Symbol(key)
+end
+
+GeneralBigDataEntry(key::Symbol,
+                    originalKey::Symbol;
+                    mimeType::String="application/octet-stream") =
+                        GeneralBigDataEntry(key, originalKey, now(), now(), mimeType)
+
+function GeneralBigDataEntry(dfg::G,
+                             var::V,
+                             originalKey::Symbol;
+                             mimeType::String="application/octet-stream") where {G <: AbstractDFG, V <: AbstractDFGVariable}
+    return GeneralBigDataEntry(uniqueKey(dfg, var, originalKey),
+                               originalKey,
+                               mimeType=mimeType)
+end
+
+# Types <: AbstractBigDataEntry
+"""
+    $(TYPEDEF)
+BigDataEntry in MongoDB.
+"""
+struct MongodbBigDataEntry <: AbstractBigDataEntry
+    key::Symbol
+    oid::NTuple{12, UInt8} #mongodb object id
+    #maybe other fields such as:
+    #flags::Bool ready, valid, locked, permissions
+    #MIMEType::Symbol
+end
+
+
+"""
+    $(TYPEDEF)
+BigDataEntry in a file.
+"""
+struct FileBigDataEntry <: AbstractBigDataEntry
+    key::Symbol
+    filename::String
+end

src/BigData/entities/DataEntries.jl

@@ -0,0 +1,25 @@
+"""
+    $(TYPEDEF)
+Simple file data store with a specified data type and a specified key type.
+"""
+struct FileDataStore <: AbstractDataStore{UInt8}
+    folder::String
+    FileDataStore(folder::String) = begin
+        if !isdir(folder)
+            @warn "Folder '$folder' doesn't exist - creating."
+             mkpath(folder)
+        end
+        return new(folder)
+    end
+end
+#
+# """
+#     $(SIGNATURES)
+# Create a file data store using a specific data type.
+# """
+# function FileDataStore(folder::String)
+#     if !isdir(folder)
+#         @warn "Folder '$folder' doesn't exist - creating."
+#     end
+#     return FileDataStore(folder)
+# end

src/BigData/entities/FileDataStore.jl

@@ -0,0 +1,24 @@
+"""
+    $(TYPEDEF)
+Simple in-memory data store with a specified data type and a specified key type.
+"""
+struct InMemoryDataStore{T, E <: AbstractBigDataEntry} <: AbstractDataStore{T}
+    data::Dict{Symbol, T}
+    entries::Dict{Symbol, E}
+end
+
+"""
+    $(SIGNATURES)
+Create an in-memory store using a specific data type.
+"""
+function InMemoryDataStore{T, E}() where {T, E <: AbstractBigDataEntry}
+    return InMemoryDataStore{T, E}(Dict{Symbol, T}(), Dict{Symbol, E}())
+end
+
+"""
+    $(SIGNATURES)
+Create an in-memory store using binary data (UInt8) as a type.
+"""
+function InMemoryDataStore()
+    return InMemoryDataStore{Vector{UInt8}, GeneralBigDataEntry}()
+end

src/BigData/entities/InMemoryDataStore.jl

@@ -0,0 +1,61 @@
+"""
+    $(SIGNATURES)
+Get the data for the specified entry, returns the data or Nothing.
+"""
+function getBigData(store::D, entry::E)::Union{Nothing, T} where {T, D <: AbstractDataStore{T}, E <: AbstractBigDataEntry}
+    error("$(typeof(store)) doesn't override 'getData'.")
+end
+
+"""
+    $(SIGNATURES)
+Adds the data to the store with the given entry. The function will warn if the entry already
+exists and will overwrite it.
+"""
+function addBigData!(store::D, entry::E, data::T)::Union{Nothing, T} where {T, D <: AbstractDataStore{T}, E <: AbstractBigDataEntry}
+    error("$(typeof(store)) doesn't override 'addData!'.")
+end
+
+"""
+    $(SIGNATURES)
+Update the data in the store. The function will error and return nothing if
+the entry does not exist.
+"""
+function updateBigData!(store::D, entry::E, data::T)::Union{Nothing, T} where {T, D <: AbstractDataStore{T}, E <: AbstractBigDataEntry}
+    error("$(typeof(store)) doesn't override 'updateData!'.")
+end
+
+"""
+    $(SIGNATURES)
+Delete the data in the store for the given entry. The function will error and return nothing if
+the entry does not exist.
+"""
+function deleteBigData!(store::D, entry::E)::Union{Nothing, T} where {T, D <: AbstractDataStore{T}, E <: AbstractBigDataEntry}
+    error("$(typeof(store)) doesn't override 'deleteData!'.")
+end
+
+"""
+    $(SIGNATURES)
+List all entries in the data store.
+"""
+function listStoreEntries(store::D)::Vector{E} where {D <: AbstractDataStore, E <: AbstractBigDataEntry}
+    error("$(typeof(store)) doesn't override 'listEntries'.")
+end
+
+"""
+    $(SIGNATURES)
+Copies all the entries from the source into the destination.
+Can specify which entries to copy with the `sourceEntries` parameter.
+Returns the list of copied entries.
+"""
+function copyStore(sourceStore::D1, destStore::D2; sourceEntries=listEntries(sourceStore))::Vector{E} where {T, D1 <: AbstractDataStore{T}, D2 <: AbstractDataStore{T}, E <: AbstractBigDataEntry}
+    # Quick check
+    destEntries = listEntries(destStore)
+    typeof(sourceEntries) != typeof(destEntries) && error("Can't copy stores, source has entries of type $(typeof(sourceEntries)), destination has entries of type $(typeof(destEntries)).")
+    # Same source/destination check
+    sourceStore == destStore && error("Can't specify same store for source and destination.")
+    # Otherwise, continue
+    for sourceEntry in sourceEntries
+        addData!(destStore, deepcopy(sourceEntry), getData(sourceStore, sourceEntry))
+    end
+    return sourceEntries
+end

src/BigData/services/AbstractDataStore.jl

@@ -0,0 +1,44 @@
+searchdir(path,key) = filter(x->occursin(key,x), readdir(path))
+filename(store::FileDataStore, entry::E) where E <: AbstractBigDataEntry = "$(store.folder)/$(entry.key).dat"
+function readentry(store::FileDataStore, entry::E) where E <: AbstractBigDataEntry
+    open(filename(store, entry)) do f
+        return read(f)
+    end
+end
+function writeentry(store::FileDataStore, entry::E, data::Vector{UInt8}) where E <: AbstractBigDataEntry
+    open(filename(store, entry), "w") do f
+        write(f, data)
+    end
+end
+
+function getBigData(store::FileDataStore, entry::E)::Union{Vector{UInt8}, Nothing} where {E <: AbstractBigDataEntry}
+    length(searchdir(store.folder, String(entry.key))) !=1 && (@warn "Could not find unique file for key '$(entry.key)'."; return nothing)
+    return readentry(store, entry)
+end
+
+function addBigData!(store::FileDataStore, entry::E, data::Vector{UInt8})::Vector{UInt8} where {E <: AbstractBigDataEntry}
+    length(searchdir(store.folder, String(entry.key))) !=0 && @warn "Key '$(entry.key)' already exists, overwriting!."
+    writeentry(store, entry, data)
+    # Update timestamp
+    entry.lastUpdatedTimestamp = now()
+    return getBigData(store, entry)
+end
+
+function updateBigData!(store::FileDataStore, entry::E, data::Vector{UInt8})::Union{Vector{UInt8}, Nothing} where {E <: AbstractBigDataEntry}
+    length(searchdir(store.folder, String(entry.key))) !=1 && (@warn "Could not find unique file for key '$(entry.key)'."; return nothing)
+    writeentry(store, entry, data)
+    # Update timestamp
+    entry.lastUpdatedTimestamp = now()
+    return getBigData(store, entry)
+end
+
+function deleteBigData!(store::FileDataStore, entry::E)::Vector{UInt8} where {E <: AbstractBigDataEntry}
+    data = getData(store, v, entry.key)
+    data == nothing && return nothing
+    rm(filename(store, entry))
+    return data
+end
+#
+# function listStoreEntries(store::FileDataStore)::Vector{E} where {E <: AbstractBigDataEntry}
+#     return collect(values(store.entries))
+# end