F# Reactive subscribe to non-standard events from COM

Question

I have a native COM library (which I can modify), and I have an F# application trying to consume an event with Rx, from an Interop library, referenced in the project.

let events = obj :?> _IInteropEvents_Event
let disposable = 
    Observable.take 1 events.ComEventArrived
    |> Observable.subscribe (fun sender data -> ()) // sender : ISender, data : IData

The error message here, which I don't fully understand, is:

The event 'ComEventArrived' has a non-standard type. If this event is declared in another CLI language, you may need to access this event using the explicit add_ComEventArrived and remove_ComEventArrived methods for the event. If this event is declared in F#, make the type of the event an instantiation of either 'IDelegateEvent<>' or 'IEvent<,_>'.

I don't mind using add_ComEventArrived but I can't figure out how to make this work with Observable

Curiously, ComEventArrived has 2 arguments of internal interop types, if I try to subscribe to other events that simply marshal an IUnknown, it works and I get no "non-standard type" error:

let events = obj :?> _ISnapshotEvents_Event
let disposable = 
    Observable.take 1 events.SnapshotEventArrived
    |> Observable.subscribe (fun sender -> ()) // sender : IUnknown (unit)

How can I do any of the following to solve the issue?

1. Modify COM library to fix the non-standard event type error.
2. Use explicit add_/remove_ functions with Observer.take/Observer.subscribe.
3. Other ways of firing an event n times before unsubscribing, without using mutable/lock.

I have read so far:

• SO: 20819860
• SO: 31896881
• SO: 41439213
• Throttling F# events

Answer 1

I don't know about COM interop specifically, but I can give an example of turning non-standard event into IObservable, which should cover your second point.

An example of non-standard event is AppDomain.CurrentDomain.AssemblyResolve where the event handler needs to return Assembly rather than returning unit. To wrap it into IObservable using the add_ and remove_ functions, you can write:

let assemblyResolve = 
  { new IObservable<_> with
      member x.Subscribe(observer) =
        let handler = ResolveEventHandler(fun o a -> 
          let mutable res = None 
          observer.OnNext((o, a, fun a -> res <- Some a))
          res.Value )
        AppDomain.CurrentDomain.add_AssemblyResolve(handler)
        { new IDisposable with
            member x.Dispose() = 
              AppDomain.CurrentDomain.remove_AssemblyResolve(handler) } }

We create a new implementation of IObservable using an F# object expression. In the Subscribe member, we create a ResolveEventHandler and add it using add_AssemblyResolve. The result of Subscribe is IDisposable implementation that then unregisters the event handler using remove_AssemblyResolve.

The ugly hack here is that the OnNext function of an observer cannot return anything, so we instead give it a three-element tuple with the arguments and also a function that sets the return value (this is specific for AssemblyResolve, so I don't suspect you'll need something like this).