Merge pull request #228 from JuliaRobotics/feature/4Q19/bigdatastores

In-memory and File-based Bigdata stores

Author: GearsAD

.gitignore

@@ -5,3 +5,4 @@ deps/deps.jl
 docs/build
 Manifest.toml
 dev
+test/sandbox.jl

src/BigData.jl

@@ -0,0 +1,17 @@
+include("entities/AbstractDataStore.jl")
+include("entities/AbstractBigDataEntries.jl")
+include("entities/InMemoryDataStore.jl")
+include("entities/FileDataStore.jl")
+
+include("services/AbstractDataStore.jl")
+include("services/AbstractBigDataEntries.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,52 @@
+"""
+    $(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
+    storeKey::Symbol # Could swap this to string, but using it as an index later, so better as a symbol I believe.
+    createdTimestamp::DateTime
+    lastUpdatedTimestamp::DateTime
+    mimeType::String
+end
+
+"""
+    $(SIGNATURES)
+Internal function to generate a unique key for the entry - userId_robotId_sessionId_variable_key.
+Simple symbol.
+"""
+function _uniqueKey(dfg::G, v::V, key::Symbol)::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, storeKey::Symbol;
+                    mimeType::String="application/octet-stream") =
+                    GeneralBigDataEntry(key, storeKey, now(), now(), mimeType)
+
+function GeneralBigDataEntry(dfg::G, var::V, key::Symbol;
+                             mimeType::String="application/octet-stream") where {G <: AbstractDFG, V <: AbstractDFGVariable}
+    return GeneralBigDataEntry(key, _uniqueKey(dfg, var, key), mimeType=mimeType)
+end
+
+"""
+    $(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/AbstractBigDataEntries.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,14 @@
+"""
+    $(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

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,17 @@
+import Base: ==
+
+function ==(a::GeneralBigDataEntry, b::GeneralBigDataEntry)
+    return a.key == b.key &&
+        a.storeKey == b.storeKey &&
+        a.mimeType == b.mimeType &&
+        Dates.value(a.createdTimestamp - b.createdTimestamp) < 1000 &&
+        Dates.value(a.lastUpdatedTimestamp - b.lastUpdatedTimestamp) < 1000 #1 second
+end
+
+function ==(a::MongodbBigDataEntry, b::MongodbBigDataEntry)
+    return a.key == b.key && a.oid == b.oid
+end
+
+function ==(a::FileBigDataEntry, b::FileBigDataEntry)
+    return a.key == b.key && a.filename == b.filename
+end

src/BigData/services/AbstractBigDataEntries.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,46 @@
+searchdir(path,key) = filter(x->occursin(key,x), readdir(path))
+filename(store::FileDataStore, entry::E) where E <: AbstractBigDataEntry = "$(store.folder)/$(entry.storeKey).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.storeKey)*".dat")) !=1 && (@warn "Could not find unique file for key '$(entry.storeKey)'."; 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.storeKey)*".dat")) !=0 && @warn "Key '$(entry.storeKey)' 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.storeKey)*".dat")) !=1 && (@warn "Could not find unique file for key '$(entry.storeKey)'."; 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 = getBigData(store, entry)
+    data == nothing && return nothing
+    rm(filename(store, entry))
+    return data
+end
+
+# TODO: Manifest file
+#
+# function listStoreEntries(store::FileDataStore)::Vector{E} where {E <: AbstractBigDataEntry}
+#     return collect(values(store.entries))
+# end