public RealtimeCursor Subscribe( 
   QueryFilter queryFilter,
   bool useRecyclingCursor
)

{
  //Note: with feature class we can also use a System Task to subscribe and
  //process rows
  // Note: call within QueuedTask.Run()
  {
    // or var rfc = realtimeDatastore.OpenTable(name) as RealtimeFeatureClass
    using RealtimeFeatureClass rfc = streamLayer.GetFeatureClass();
    //non-recycling cursor - 2nd param "false"
    //subscribe, pre-existing rows are not searched
    using RealtimeCursor rc = rfc.Subscribe(qfilter, false);
    SpatialQueryFilter spatialFilter = new SpatialQueryFilter();
    //waiting for new features to be streamed
    //default is no cancellation
    while (rc.WaitForRowsAsync().Result)
    {
      while (rc.MoveNext())
      {
        using (var row = rc.Current)
        {
          switch (row.GetRowSource())
          {
            case RealtimeRowSource.EventInsert:
              //getting geometry from new events as they arrive
              Polygon poly = ((RealtimeFeature)row).GetShape() as Polygon;

              //using the geometry to select features from another feature layer
              spatialFilter.FilterGeometry = poly;//project poly if needed...
              featureLayer.Select(spatialFilter);
              continue;
            default:
              continue;
          }
        }
      }
    }
    //row cursor is disposed. row cursor is unsubscribed
  }
}

{
  //Note we can use System Task with the Realtime feature class
  //for subscribe

  // Note: call within QueuedTask.Run()
  {
    using RealtimeFeatureClass rfc = streamLayer.GetFeatureClass();
    //non-recycling cursor - 2nd param "false"
    using RealtimeCursor rc = rfc.SearchAndSubscribe(qfilter, false);

    //waiting for new features to be streamed
    //default is no cancellation - use await in async method
    while (rc.WaitForRowsAsync().Result)
    {
      //pre-existing rows will be retrieved that were searched
      while (rc.MoveNext())
      {
        using RealtimeRow row = rc.Current;
        var row_source = row.GetRowSource();
        switch (row_source)
        {
          case RealtimeRowSource.EventDelete:
            //TODO - handle deletes
            break;
          case RealtimeRowSource.EventInsert:
            //TODO handle inserts
            break;
          case RealtimeRowSource.PreExisting:
            //TODO handle pre-existing rows
            break;
        }
      }
    }
    //row cursor is disposed. row cursor is unsubscribed
  }
}

{
  // Note: call within QueuedTask.Run()
  {
    using RealtimeFeatureClass rfc = streamLayer.GetFeatureClass();
    //Recycling cursor - 2nd param "true"
    using RealtimeCursor rc = rfc.SearchAndSubscribe(qfilter, true);
    //auto-cancel after 20 seconds
    var cancel = new CancellationTokenSource(new TimeSpan(0, 0, 20));
    //catch TaskCanceledException
    try
    {
      // Use await in async method
      while (rc.WaitForRowsAsync(cancel.Token).Result)
      {
        //check for row events
        while (rc.MoveNext())
        {
          using RealtimeRow record = rc.Current;
          // Process the record
        }
      }
    }
    catch (TaskCanceledException)
    {
      //Handle cancellation as needed
    }
    cancel.Dispose();
  }
}