@@ -27,17 +27,24 @@ enum DatabaseError: Error {
2727 }
2828}
2929
30+
31+ func subscription_cb( ctx: UnsafeMutableRawPointer ? , subid: UInt64 ) -> Void {
32+ guard let ctx else { return }
33+ let ndb = Unmanaged < Ndb > . fromOpaque ( ctx) . takeUnretainedValue ( )
34+ ndb. sub_cb ? ( subid)
35+ }
36+
3037class Ndb {
3138 var ndb : ndb_t
3239 let path : String ?
3340 let owns_db : Bool
3441 var generation : Int
42+ let sub_cb : ( ( UInt64 ) -> ( ) ) ?
3543 private var closed : Bool
3644
3745 var is_closed : Bool {
3846 self . closed || self . ndb. ndb == nil
3947 }
40-
4148 static func safemode( ) -> Ndb ? {
4249 guard let path = db_path ?? old_db_path else { return nil }
4350
@@ -49,7 +56,8 @@ class Ndb {
4956 }
5057 }
5158
52- guard let ndb = Ndb ( path: path) else {
59+ let ndb = Ndb ( path: path)
60+ guard let _ = ndb. open ( ) else {
5361 return nil
5462 }
5563
@@ -79,14 +87,14 @@ class Ndb {
7987 print ( " txn: NOSTRDB EMPTY " )
8088 return Ndb ( ndb: ndb_t ( ndb: nil ) )
8189 }
82-
83- static func open( path : String ? = nil , owns_db_file : Bool = true ) -> ndb_t ? {
90+
91+ func open( ) -> ndb_t ? {
8492 var ndb_p : OpaquePointer ? = nil
8593
8694 let ingest_threads : Int32 = 4
8795 var mapsize : Int = 1024 * 1024 * 1024 * 32
8896
89- if path == nil && owns_db_file {
97+ if path == nil && owns_db {
9098 // `nil` path indicates the default path will be used.
9199 // The default path changed over time, so migrate the database to the new location if needed
92100 do {
@@ -99,7 +107,7 @@ class Ndb {
99107 }
100108
101109 guard let db_path = Self . db_path,
102- owns_db_file || Self . db_files_exist ( path: db_path) else {
110+ owns_db || Self . db_files_exist ( path: db_path) else {
103111 return nil // If the caller claims to not own the DB file, and the DB files do not exist, then we should not initialize Ndb
104112 }
105113
@@ -109,8 +117,9 @@ class Ndb {
109117
110118 let ok = path. withCString { testdir in
111119 var ok = false
120+ let ctx = Unmanaged . passUnretained ( self ) . toOpaque ( )
112121 while !ok && mapsize > 1024 * 1024 * 700 {
113- var cfg = ndb_config ( flags: 0 , ingester_threads: ingest_threads, mapsize: mapsize, filter_context: nil , ingest_filter: nil , sub_cb_ctx: nil , sub_cb: nil )
122+ var cfg = ndb_config ( flags: 0 , ingester_threads: ingest_threads, mapsize: mapsize, filter_context: nil , ingest_filter: nil , sub_cb_ctx: ctx , sub_cb: subscription_cb )
114123 ok = ndb_init ( & ndb_p, testdir, & cfg) != 0
115124 if !ok {
116125 mapsize /= 2
@@ -123,19 +132,17 @@ class Ndb {
123132 return nil
124133 }
125134
126- return ndb_t ( ndb: ndb_p)
135+ self . ndb = ndb_t ( ndb: ndb_p)
136+ return self . ndb
127137 }
128138
129- init ? ( path: String ? = nil , owns_db_file: Bool = true ) {
130- guard let db = Self . open ( path: path, owns_db_file: owns_db_file) else {
131- return nil
132- }
133-
139+ init ( path: String ? = nil , owns_db_file: Bool = true , sub_cb: ( ( UInt64 ) -> ( ) ) ? = nil ) {
134140 self . generation = 0
135141 self . path = path
136142 self . owns_db = owns_db_file
137- self . ndb = db
138143 self . closed = false
144+ self . sub_cb = sub_cb
145+ self . ndb = ndb_t ( )
139146 }
140147
141148 private static func migrate_db_location_if_needed( ) throws {
@@ -181,6 +188,7 @@ class Ndb {
181188 self . path = nil
182189 self . owns_db = true
183190 self . closed = false
191+ self . sub_cb = nil
184192 }
185193
186194 func close( ) {
@@ -193,8 +201,8 @@ class Ndb {
193201 }
194202
195203 func reopen( ) -> Bool {
196- guard self . is_closed ,
197- let db = Self . open ( path : self . path , owns_db_file : self . owns_db ) else {
204+ let ctx = Unmanaged . passUnretained ( self ) . toOpaque ( )
205+ guard self . is_closed , let db = self . open ( ) else {
198206 return false
199207 }
200208
@@ -205,6 +213,20 @@ class Ndb {
205213 return true
206214 }
207215
216+ func poll_for_notes( subid: Int64 , capacity: Int ) -> [ NoteKey ] {
217+ var buf = Array< UInt64> . init( repeating: 0 , count: capacity)
218+
219+ let r = buf. withUnsafeMutableBufferPointer { bytes in
220+ return ndb_poll_for_notes ( self . ndb. ndb, UInt64 ( subid) , bytes. baseAddress, Int32 ( capacity) )
221+ }
222+
223+ guard r != 0 else {
224+ return [ ]
225+ }
226+
227+ return Array ( buf. prefix ( Int ( r) ) )
228+ }
229+
208230 func lookup_blocks_by_key_with_txn< Y> ( _ key: NoteKey , txn: NdbTxn < Y > ) -> NdbBlocks ? {
209231 guard let blocks = ndb_get_blocks_by_key ( self . ndb. ndb, & txn. txn, key) else {
210232 return nil
0 commit comments