Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 21 additions & 5 deletions nimutils/sinks.nim
Original file line number Diff line number Diff line change
Expand Up @@ -255,6 +255,7 @@ type S3SinkState* = ref object of RootRef
objPath*: string
nameBase*: string
extra*: string
endpoint*: string

proc s3SinkInit(cfg: SinkConfig): bool =
try:
Expand All @@ -266,21 +267,31 @@ proc s3SinkInit(cfg: SinkConfig): bool =
token = cfg.params.getOrDefault("token", "")
region = cfg.params.getOrDefault("region", defaultRegion)
extra = cfg.params.getOrDefault("extra", "")
baseObj = uri.path[1 .. ^1] # Strip the leading /
# uri.path is "" when no path is given (e.g. s3://bucket); guard
# against that before slicing off the leading '/'.
rawPath = uri.path
baseObj = if rawPath.len > 1: rawPath[1 .. ^1] else: ""
(objPath, nameBase) = splitPath(baseObj)
endpoint = cfg.params.getOrDefault("endpoint", "")

cfg.private = S3SinkState(region: region, uri: uri, uid: uid,
secret: secret, token: token,
bucket: bucket, objPath: objPath,
nameBase: nameBase, extra: extra)
nameBase: nameBase, extra: extra,
endpoint: endpoint)
return true
except:
return false

proc s3SinkOut(msg: string, cfg: SinkConfig, t: Topic, ignored: StringTable) =
var
state = S3SinkState(cfg.private)
client = newS3Client((state.uid, state.secret, state.token), state.region)
client = if state.endpoint != "":
newS3Client((state.uid, state.secret, state.token),
state.region, state.endpoint)
else:
newS3Client((state.uid, state.secret, state.token),
state.region)

cfg.iolog(t, "Open") # Not really a connect...

Expand All @@ -296,7 +307,11 @@ proc s3SinkOut(msg: string, cfg: SinkConfig, t: Topic, ignored: StringTable) =

let
newTail = objParts.join("-")
newPath = joinPath(state.objPath, newTail)
rawPath = joinPath(state.objPath, newTail)
# put_object expects an absolute path; joinPath doesn't add a leading
# slash when objPath is empty, but the awsclient URL template requires
# one to separate the bucket name from the object key.
newPath = if rawPath.startsWith("/"): rawPath else: "/" & rawPath
response = client.put_object(state.bucket, newPath, msg)

cfg.iolog(t, "Post to: " & newPath & "; response = " & response.status)
Expand Down Expand Up @@ -469,7 +484,8 @@ proc addS3Sink*() =
"token" : false,
"uri" : true,
"region" : false,
"extra" : false
"extra" : false,
"endpoint" : false
}.toTable()

record.initFunction = some(InitCallback(s3SinkInit))
Expand Down
Loading