I have some code written in F#, using .NET Core and ASP.NET Core. It's supposed to register change listeners to the database, and it does work occasionally:
namespace Raptor.ChangeListener.Oracle
open System
open System.Data
open Microsoft.Extensions.Logging
open Microsoft.Extensions.Options
open Oracle.ManagedDataAccess.Client
open Raptor.ChangeListener
[<Sealed>]
type OracleChangeListenerOptions() =
member val ConnectionString = Unchecked.defaultof<string> with get, set
member val Query = Unchecked.defaultof<string> with get, set
member val Timeout = Unchecked.defaultof<TimeSpan> with get, set
type OracleChangeListener(options: OracleChangeListenerOptions IOptions, logger: ILogger<OracleChangeListener>) =
let options = options.Value
let mutable disposed = false
let sync = obj()
let mutable handlers: (Guid * Action) list = []
let mutable connection: OracleConnection = null
let mutable command : OracleCommand = null
let mutable dependency: OracleDependency = null
let startup =
async {
connection <- new OracleConnection(options.ConnectionString)
connection.StateChange.Add(fun state ->
if not disposed then
match state.CurrentState with
| ConnectionState.Broken -> logger.LogError(sprintf "Connection Broken: %s" options.Query)
| ConnectionState.Closed -> logger.LogError(sprintf "Connection Closed: %s" options.Query)
| _ -> ()
)
do! connection.OpenAsync() |> Async.AwaitTask
command <- new OracleCommand(options.Query, connection)
dependency <- new OracleDependency(command)
// these don't exist without first setting the dependency
command.Notification.IsNotifiedOnce <- false
command.Notification.Timeout <- int64 options.Timeout.TotalSeconds
let test = (command, dependency)
dependency.OnChange.Add(fun state ->
if not disposed then
try
logger.LogInformation(
let info = state.Info.ToString()
let typ = state.Type.ToString()
sprintf "%s Notification Change Type: %s, TableName: %A" info typ state.ResourceNames
)
for _, handler in handlers do
try
handler.Invoke()
with ex ->
logger.LogError(ex, "OracleChangeListener handler error")
with ex ->
logger.LogError(ex, "OracleChangeListener error")
)
match command.ExecuteNonQuery() with
| -1 -> ()
| n -> failwithf "Error Code In Listener: %i" n
}
|> Async.StartAsTask
interface IChangeListener with
member _.AddHandler handler =
lock sync (fun () ->
let key = Guid.NewGuid()
handlers <- List.append handlers [(key, handler)]
key
)
member _.RemoveHandler key =
lock sync (fun () ->
handlers <- handlers |> List.where (fun (k, _) -> k <> key)
)
interface IDisposable with
member this.Dispose() =
if not disposed then
disposed <- true // flag early for internal checks
if not (isNull dependency) then dependency.RemoveRegistration(connection)
if not (isNull command ) then command .Dispose()
if not (isNull connection) then connection.Dispose()
GC.SuppressFinalize(this)
[<Sealed>]
type OracleChangeListener<'TCategory>(options, logger) =
inherit OracleChangeListener(options, logger)
interface IChangeListener<'TCategory>
when I select from DBA_CHANGE_NOTIFICATION_REGS in the database, I see all the proper registrations in place:
But more often than not, and I can't determine what the cause of this is, the registrations are made with a port of -1 instead of a valid port number, like so:
When this happens, none of the services are properly notified when registered tables are changed. Additionally, all invalid rows are cleared when a change that ought to trigger a notification occurs.