Live durability: do-block, frame rotation, writer wrapper

Project.toml

@@ -1,6 +1,6 @@
 name = "DatabentoAPI"
 uuid = "feb0aed8-3291-460b-9688-521dcb17a4bf"
-version = "0.1.1"
+version = "0.1.2"
 authors = ["Tyler Beason <tbeas12@gmail.com>"]
 
 [deps]
@@ -22,7 +22,7 @@ URIs = "5c2747f8-b7ea-4ff2-ba2e-563bfd36b1d4"
 [compat]
 CodecZstd = "0.8"
 DataFrames = "1.7"
-DatabentoBinaryEncoding = "0.1"
+DatabentoBinaryEncoding = "0.1.1"
 EnumX = "1.0"
 HTTP = "1.10"
 JSON3 = "1.14"

README.md

@@ -99,32 +99,72 @@ resolve, foreach_record
 
 ## Live example
 
+The do-block form mirrors `Base.open` and guarantees `close(client)` runs on
+`Ctrl-C`, exceptions, or normal exit — no `try/finally` boilerplate needed.
+
 ```julia
 using DatabentoAPI
 
-client = Live(dataset = "OPRA.PILLAR")
-connect!(client)
-subscribe!(client;
-    schema   = Schema.TRADES,
-    symbols  = ["AAPL"],
-    stype_in = SType.RAW_SYMBOL)
-start!(client)
+Live(dataset = "OPRA.PILLAR") do client
+    connect!(client)
+    subscribe!(client;
+        schema   = Schema.TRADES,
+        symbols  = ["AAPL"],
+        stype_in = SType.RAW_SYMBOL)
+    start!(client)
+    for rec in client                     # iterator pulls from internal Channel
+        println(rec)
+        rec isa DBN.TradeMsg && rec.price > some_threshold && break
+    end
+end   # Ctrl-C here triggers a clean close(client)
+```
 
-for rec in client                         # iterator pulls from internal Channel
-    println(rec)
-    rec isa DBN.TradeMsg && rec.price > some_threshold && break
-end
+Manual lifecycle still works if you prefer:
+
+```julia
+client = Live(dataset = "OPRA.PILLAR")
+connect!(client); subscribe!(client; …); start!(client)
+for rec in client; …; end
 close(client)
 ```
 
 Callback variant:
 
 ```julia
-subscribe_callback(client, rec -> @info "tick" rec)
-sleep(30)
-close(client)
+Live(dataset = "OPRA.PILLAR") do client
+    connect!(client)
+    subscribe!(client; schema = Schema.TRADES, symbols = ["AAPL"], stype_in = SType.RAW_SYMBOL)
+    subscribe_callback(client, rec -> @info "tick" rec)
+    start!(client)
+    sleep(30)
+end
+```
+
+### Writing records to disk yourself
+
+If you want to subscribe and write records to a DBN file from your own loop —
+including in-file zstd frame rotation for crash safety — pair `Live` with
+`open_dbn_writer`:
+
+```julia
+Live(dataset = "GLBX.MDP3") do client
+    connect!(client)
+    subscribe!(client; schema = Schema.TRADES, symbols = ["ES.FUT"], stype_in = SType.PARENT)
+    start!(client)
+    open_dbn_writer(; base_dir = "./capture", dataset = "GLBX.MDP3",
+                      schema = Schema.TRADES, symbols = ["ES.FUT"],
+                      stype_in = SType.PARENT,
+                      frame_seconds = 60.0) do writer
+        for rec in client
+            write_record!(writer, rec)
+        end
+    end
+end
 ```
 
+For the common case of "subscribe and dump every record to disk", use
+[`stream_to_file`](@ref) directly — it wraps the same pieces.
+
 ### Typed-channel variant (`typed = true`)
 
 For high-throughput consumers, opt into per-schema typed channels:

src/DatabentoAPI.jl

@@ -109,5 +109,6 @@ export channel, control_channel
 
 # Live capture to file
 export stream_to_file, stream_multi_to_files, default_live_path, read_capture
+export open_dbn_writer, write_record!
 
 end # module DatabentoAPI

src/live/client.jl

@@ -119,6 +119,38 @@ function Base.show(io::IO, c::Live)
               ", connected=", c.connected, ", started=", c.started, ")")
 end
 
+"""
+    Live(f::Function, args...; kwargs...)
+
+Do-block form mirroring `Base.open(f, path)`. Constructs the client, runs
+`f(client)`, and guarantees `close(client)` in `finally` — so on
+`InterruptException` (Ctrl-C), an exception, or normal exit the socket and
+channels are torn down without the caller writing the `try/finally`
+themselves.
+
+```julia
+Live(dataset = "GLBX.MDP3") do client
+    connect!(client)
+    subscribe!(client; schema = Schema.TRADES, symbols = ["ES.FUT"], stype_in = SType.PARENT)
+    start!(client)
+    for rec in client
+        # … your code …
+    end
+end  # Ctrl-C here triggers a clean close(client)
+```
+
+The manual lifecycle (`Live(...)` → `connect!` → ... → `close(client)`) keeps
+working unchanged — this is purely additive.
+"""
+function Live(f::Function, args...; kwargs...)
+    client = Live(args...; kwargs...)
+    try
+        return f(client)
+    finally
+        try; close(client); catch; end
+    end
+end
+
 """
     control_channel(client::Live) -> Channel{DBN.DBNRecord}
 
@@ -189,6 +221,9 @@ channel plus the control channel in typed mode).
 """
 function Base.close(c::Live)
     c.closed && return
+    # Flip the flag BEFORE any teardown so re-entry (e.g. user pressed
+    # Ctrl-C twice, or close() is called from a finally inside f()) is a
+    # no-op and doesn't double-close anything.
     c.closed = true
     try
         if c.connected && c.socket !== nothing && isopen(c.socket)
@@ -204,24 +239,24 @@ function Base.close(c::Live)
     catch
     end
     # Close every channel we own. Each in its own try so a single bad close
-    # doesn't leave others stranded.
+    # doesn't leave others stranded. Iteration runs regardless of c.typed
+    # because untyped clients have an empty typed_data_channels dict and a
+    # nothing control_channel, so the typed branches are no-ops.
     try
         c.channel === nothing || (isopen(c.channel) && close(c.channel))
     catch
     end
-    if c.typed
-        for ch in values(c.typed_data_channels)
-            try
-                isopen(ch) && close(ch)
-            catch
-            end
-        end
+    for ch in values(c.typed_data_channels)
         try
-            c.control_channel === nothing ||
-                (isopen(c.control_channel) && close(c.control_channel))
+            isopen(ch) && close(ch)
         catch
         end
     end
+    try
+        c.control_channel === nothing ||
+            (isopen(c.control_channel) && close(c.control_channel))
+    catch
+    end
     return nothing
 end