add eventSink
This commit is contained in:
+1
-1
@@ -1,6 +1,6 @@
|
|||||||
name = "GeneralUtils"
|
name = "GeneralUtils"
|
||||||
uuid = "c6c72f09-b708-4ac8-ac7c-2084d70108fe"
|
uuid = "c6c72f09-b708-4ac8-ac7c-2084d70108fe"
|
||||||
version = "0.6.6"
|
version = "0.6.7"
|
||||||
authors = ["tonaerospace <tonaerospace.etc@gmail.com>"]
|
authors = ["tonaerospace <tonaerospace.etc@gmail.com>"]
|
||||||
|
|
||||||
[deps]
|
[deps]
|
||||||
|
|||||||
+91
-45
@@ -4,7 +4,7 @@ export timedifference, showstracktrace, findHighestIndexKey, uuid4snakecase, rep
|
|||||||
findMatchingDictKey, randstring, randstrings, timeout,
|
findMatchingDictKey, randstring, randstrings, timeout,
|
||||||
dataframeToCSV, dfToVectorDict, disintegrate_vectorDict, getDataFrameValue, dfRowtoString,
|
dataframeToCSV, dfToVectorDict, disintegrate_vectorDict, getDataFrameValue, dfRowtoString,
|
||||||
dfToString, dataframe_to_json_list, dictToString, dictToString_noKey, issomething,
|
dfToString, dataframe_to_json_list, dictToString, dictToString_noKey, issomething,
|
||||||
dictToString_numbering, extract_triple_backtick_text, logger,
|
dictToString_numbering, extract_triple_backtick_text, logger, eventSink,
|
||||||
countGivenWords, remove_french_accents, removestring,
|
countGivenWords, remove_french_accents, removestring,
|
||||||
extractTextBetweenCharacter, extractTextBetweenString,
|
extractTextBetweenCharacter, extractTextBetweenString,
|
||||||
convertCamelSnakeKebabCase, fitrange, recentElementsIndex, nonRecentElementsIndex
|
convertCamelSnakeKebabCase, fitrange, recentElementsIndex, nonRecentElementsIndex
|
||||||
@@ -1174,64 +1174,110 @@ end
|
|||||||
|
|
||||||
|
|
||||||
"""
|
"""
|
||||||
Write a log entry to a local markdown file and optionally publish it to a NATS status topic.
|
Create an event sink for logging messages and/or publishing to NATS or printing to console. File logging defaults to false.
|
||||||
Each entry is prefixed with a timestamp and service name: `-[YYYY-MM-DDTHH:MM:SS.sss] [serviceName] msg`.
|
|
||||||
The log file is capped at 100 entries (oldest entries are truncated on rotation).
|
|
||||||
|
|
||||||
# Arguments
|
# Fields
|
||||||
- `msg::String`: Log message body.
|
- `natsConn`: NATS connection, or `nothing` for file-only sink
|
||||||
- `service_name::String`: Service identifier, inserted as `[name]` prefix. Defaults to `config["thisServiceName"]`.
|
- `topic`: NATS topic to publish to, or `nothing` for file-only sink
|
||||||
- `nats_conn`: Optional NATS connection handle. Pass to enable NATS publishing.
|
- `service_name`: Optional prefix added to log entries
|
||||||
- `nats_status_topic`: Optional NATS topic to publish log entries to.
|
- `log_file`: Path to the log file
|
||||||
- `log_dir::String`: Directory for the log file. Defaults to `"./log"`.
|
- `max_log_entries`: Maximum number of entries to keep in the log file
|
||||||
- `log_file::String`: Full path to the markdown log file. Defaults to `"./log/log.md"`.
|
|
||||||
- `max_log_entries::Int`: Maximum number of log entries to keep in the file before rotating. Defaults to `100`.
|
# Callable interface
|
||||||
|
```julia
|
||||||
|
(es::eventSink)(msg::String; log=false, publish=true, console=true) -> Nothing
|
||||||
|
```
|
||||||
|
|
||||||
|
Three independent outputs — each defaults to `true` except `log`:
|
||||||
|
- `console` (default `true`) → println to stdout
|
||||||
|
- `publish` (default `true`) → publish to NATS topic (no-op if no connection)
|
||||||
|
- `log` (default `false`) → append to `log_file`
|
||||||
|
|
||||||
|
# Examples
|
||||||
|
```julia
|
||||||
|
using NATS, Dates
|
||||||
|
|
||||||
|
conn = NATS.connect("nats.yiem.cc")
|
||||||
|
|
||||||
|
# ——— With NATS (console + NATS by default) ———
|
||||||
|
es = eventSink(natsConn=conn, topic="agent.events", service_name="agent1")
|
||||||
|
es("starting up") # console + NATS
|
||||||
|
es("logged"; log=true) # console + file + NATS
|
||||||
|
es("quiet"; console=false) # NATS only
|
||||||
|
es("file+nats"; log=true, console=false) # file + NATS
|
||||||
|
es("console-only"; publish=false, log=false) # console only
|
||||||
|
|
||||||
|
# ——— File-only sink (no NATS connection) ———
|
||||||
|
es2 = eventSink(log_file="./log/eventSink2.md", service_name="worker")
|
||||||
|
es2("ready") # console only
|
||||||
|
es2("file-logged"; log=true) # console + file
|
||||||
|
es2("file-only"; log=true, console=false) # file only
|
||||||
|
es2("silent"; console=false) # nothing (no NATS, log=false, console=false)
|
||||||
|
|
||||||
|
NATS.drain(conn)
|
||||||
|
```
|
||||||
"""
|
"""
|
||||||
function logger(msg::String; service_name::String=config["thisServiceName"],
|
struct eventSink
|
||||||
nats_conn=nothing, nats_status_topic=nothing, log_dir::String="./log",
|
natsConn::Union{NATS.Connection, Nothing}
|
||||||
log_file::String=joinpath(log_dir, "log.md"), max_log_entries::Int=100)::Nothing
|
topic::Union{String, Nothing}
|
||||||
|
service_name::String
|
||||||
|
log_file::String
|
||||||
|
max_log_entries::Int
|
||||||
|
end
|
||||||
|
|
||||||
|
eventSink(; natsConn=nothing, topic=nothing, service_name="", log_file="./log/eventSink.md", max_log_entries=100) =
|
||||||
|
eventSink(natsConn, topic, service_name, log_file, max_log_entries)
|
||||||
|
|
||||||
|
"""
|
||||||
|
rotate!(es::eventSink) -> Nothing
|
||||||
|
|
||||||
|
Rewrite the log file keeping only the last `max_log_entries` entries.
|
||||||
|
Call after writing when the entry count reaches the limit.
|
||||||
|
"""
|
||||||
|
function rotate!(es::eventSink)
|
||||||
|
content = read(es.log_file, String)
|
||||||
|
entries = split(content, r"- \[", keepempty=false)
|
||||||
|
open(es.log_file, "w") do f
|
||||||
|
write(f, "- [")
|
||||||
|
for e in entries[2:end]
|
||||||
|
write(f, "- [", e)
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
function (es::eventSink)(msg::String; log=false, publish=true, console=true)
|
||||||
timestamp = Dates.format(now(), "yyyy-mm-ddTHH:MM:SS.sss")
|
timestamp = Dates.format(now(), "yyyy-mm-ddTHH:MM:SS.sss")
|
||||||
svc_prefix = isempty(service_name) ? "" : "[$service_name] "
|
svc_prefix = isempty(es.service_name) ? "" : "[$(es.service_name)] "
|
||||||
entry = "- [$timestamp] $svc_prefix$msg\n"
|
entry = "- [$timestamp] $svc_prefix$msg\n"
|
||||||
mkpath(log_dir)
|
if console
|
||||||
lk = ReentrantLock()
|
println("[$timestamp] $svc_prefix$msg")
|
||||||
lock(lk) do
|
end
|
||||||
if isfile(log_file)
|
if log
|
||||||
content = read(log_file, String)
|
log_dir = dirname(es.log_file)
|
||||||
entries = split(content, r"- \[", keepempty=false)
|
mkpath(log_dir)
|
||||||
if length(entries) >= max_log_entries
|
open(es.log_file, "a") do f
|
||||||
open(log_file, "w") do f
|
write(f, entry)
|
||||||
write(f, "- [")
|
end
|
||||||
for e in entries[2:end]
|
# Rotate after write — keep only last max_log_entries entries
|
||||||
write(f, "- [", e)
|
content = read(es.log_file, String)
|
||||||
end
|
entries = split(content, r"- \[", keepempty=false)
|
||||||
write(f, entry)
|
if length(entries) > es.max_log_entries
|
||||||
end
|
rotate!(es)
|
||||||
else
|
|
||||||
open(log_file, "a") do f
|
|
||||||
write(f, entry)
|
|
||||||
end
|
|
||||||
end
|
|
||||||
else
|
|
||||||
open(log_file, "w") do f
|
|
||||||
write(f, entry)
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
println("\n", entry)
|
if publish && es.topic !== nothing && es.natsConn !== nothing
|
||||||
if nats_conn !== nothing && nats_status_topic !== nothing
|
|
||||||
try
|
try
|
||||||
NATS.publish(nats_conn, nats_status_topic, entry)
|
NATS.publish(es.natsConn, es.topic, entry)
|
||||||
catch e
|
catch e
|
||||||
@error "log NATS publish failed" exception=e
|
@warn "eventSink NATS publish failed" exception=e
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
return nothing
|
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
end # module util
|
end # module util
|
||||||
Reference in New Issue
Block a user