public RealtimeCursor Subscribe( 
   QueryFilter queryFilter,
   bool useRecyclingCursor
)

//Note: with feature class we can also use a System Task to subscribe and
//process rows
await QueuedTask.Run(async () =>
{
  // or var rfc = realtimeDatastore.OpenTable(name) as RealtimeFeatureClass
  using (var rfc = streamLayer.GetFeatureClass())
  {
    //non-recycling cursor - 2nd param "false"
    //subscribe, pre-existing rows are not searched
    using (var rc = rfc.Subscribe(qfilter, false))
    {
      SpatialQueryFilter spatialFilter = new SpatialQueryFilter();
      //waiting for new features to be streamed
      //default is no cancellation
      while (await rc.WaitForRowsAsync())
      {
        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...
                countyFeatureLayer.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
await System.Threading.Tasks.Task.Run(async () =>
// or use ... QueuedTask.Run()
{
  using (var rfc = streamLayer.GetFeatureClass())
  {
    //non-recycling cursor - 2nd param "false"
    using (var rc = rfc.SearchAndSubscribe(qfilter, false))
    {
      //waiting for new features to be streamed
      //default is no cancellation
      while (await rc.WaitForRowsAsync())
      {
        //pre-existing rows will be retrieved that were searched
        while (rc.MoveNext())
        {
          using (var 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
  }
});

await QueuedTask.Run(async () =>
{
  //Recycling cursor - 2nd param "true"
  //or streamLayer.Subscribe(qfilter, true) to just subscribe
  using (var rc = streamLayer.SearchAndSubscribe(qfilter, true))
  {
    //auto-cancel after 20 seconds
    var cancel = new CancellationTokenSource(new TimeSpan(0, 0, 20));
    //catch TaskCanceledException
    try
    {
      while (await rc.WaitForRowsAsync(cancel.Token))
      {
        //check for row events
        while (rc.MoveNext())
        {
          using (var row = rc.Current)
          {
            //etc
          }
        }
      }
    }
    catch (TaskCanceledException )
    {
      //Handle cancellation as needed
    }
    cancel.Dispose();
  }
});