public Task<bool> WaitForRowsAsync( 
   CancellationToken cancellationToken
)

{
  await QueuedTask.Run(async () =>
  {
    //and assuming you have established a connection to a knowledge graph
    //...
    //Construct an openCypher query - return the first 10 entities (whatever
    //they are...)
    var query = "MATCH (n) RETURN n LIMIT 10";//default limit is 100 if not specified
                                              //other examples...
                                              //query = "MATCH (a:Person) RETURN [a.name, a.age] ORDER BY a.age DESC LIMIT 50";
                                              //query = "MATCH (b:Person) RETURN { Xperson: { Xname: b.name, Xage: b.age } } ORDER BY b.name DESC";
                                              //query = "MATCH p = (c:Person)-[:HasCar]-() RETURN p ORDER BY c.name DESC";

    //Create a query filter
    //Note: OutputSpatialReference is currently ignored
    var kg_qf = new KnowledgeGraphQueryFilter()
    {
      QueryText = query
    };
    //Optionally - u can choose to include provenance in the results
    //(_if_ the KG has provenance - otherwise the query will fail)
    bool includeProvenanceIfPresent = KnowledgeGraphSupportsProvenance(kg);

    if (includeProvenanceIfPresent)
    {
      //Only include if the KG has provenance
      kg_qf.ProvenanceBehavior =
          KnowledgeGraphProvenanceBehavior.Include;//default is exclude
    }
    //submit the query - returns a KnowledgeGraphCursor
    using (var kg_rc = kg.SubmitQuery(kg_qf))
    {
      //wait for rows to be returned from the server
      //note the "await"...
      while (await kg_rc.WaitForRowsAsync())
      {
        //Rows have been retrieved - process this "batch"...
        while (kg_rc.MoveNext())
        {
          //Get the current KnowledgeGraphRow
          using (var graph_row = kg_rc.Current)
          {
            //Graph row is an array, process all returned values...
            var val_count = (int)graph_row.GetCount();
            for (int i = 0; i < val_count; i++)
            {
              var retval = graph_row[i];
              //Process row value (note: recursive)
              //See "Process a KnowledgeGraphRow Value" snippet
              ProcessKnowledgeGraphRowValue(retval);
            }
          }
        }
      }//WaitForRowsAsync
    }//SubmitQuery
  });
}

{
  // Note: call within QueuedTask.Run()
  {
    //query filter can be null to search and retrieve all rows
    //true means recycling cursor
    using (var rc = streamLayer.SearchAndSubscribe(qfilter, true))
    {
      //waiting for new features to be streamed
      //default is no cancellation
      while (rc.WaitForRowsAsync().Result)
      {
        while (rc.MoveNext())
        {
          using (var row = rc.Current)
          {
            //determine the origin of the row event
            switch (row.GetRowSource())
            {
              case RealtimeRowSource.PreExisting:
                //pre-existing row at the time of subscribe
                continue;
              case RealtimeRowSource.EventInsert:
                //row was inserted after subscribe
                continue;
              case RealtimeRowSource.EventDelete:
                //row was deleted after subscribe
                continue;
            }
          }
        }
      }
    }//row cursor is disposed. row cursor is unsubscribed

    //....or....
    //Use the feature class instead of the layer
    using var 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
      while (rc.WaitForRowsAsync().Result)
      {
        //etc
      }
    }
  }
}

{
  // Note: call within QueuedTask.Run()
  {
    //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 (rc.WaitForRowsAsync(cancel.Token).Result)
        {
          //check for row events
          while (rc.MoveNext())
          {
            using var row = rc.Current;
            //etc
          }
        }
      }
      catch (TaskCanceledException)
      {
        //Handle cancellation as needed
      }
      cancel.Dispose();
    }
  }
}

{
  //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
  }
}

{
  //On the QueuedTask...
  //and assuming you have established a connection to a knowledge graph
  //...
  //submit query or search to return a KnowledgeGraphCursor
  //using (var kgRowCursor = kg.SubmitQuery(kg_qf)) {
  //using (var kgRowCursor = kg.SubmitSearch(kg_sf)) {
  //...
  //wait for rows to be returned from the server
  //"auto-cancel" after 20 seconds
  var cancel = new CancellationTokenSource(new TimeSpan(0, 0, 20));
  //catch TaskCanceledException
  try
  {
    while (await kgRowCursor.WaitForRowsAsync(cancel.Token))
    {
      //check for row events
      while (kgRowCursor.MoveNext())
      {
        using (var graph_row = kgRowCursor.Current)
        {
          //Graph row is an array, process all returned values...
          var val_count = (int)graph_row.GetCount();
          for (int i = 0; i < val_count; i++)
          {
            var retval = graph_row[i];
            //Process row value (note: recursive)
            //See "Process a KnowledgeGraphRow Value" snippet
            ProcessKnowledgeGraphRowValue(retval);
          }
        }
      }
    }
  }
  //Timeout expired
  catch (TaskCanceledException tce)
  {
    //Handle cancellation as needed
  }
  cancel.Dispose();
}

{
  RealtimeCursor rc = null;
  bool SomeConditionForCancel = false;

  //somewhere in our code we create a CancellationTokenSource
  var cancel = new CancellationTokenSource();
  //...

  //call cancel on the CancellationTokenSource anywhere in
  //the add-in, assuming the CancellationTokenSource is in scope
  if (SomeConditionForCancel)
    cancel.Cancel();//<-- will cancel the token

  //Within QueuedTask we are subscribed! streamLayer.Subscribe() or SearchAndSubscribe()
  try
  {
    //TaskCanceledException will be thrown when the token is cancelled
    while (rc.WaitForRowsAsync(cancel.Token).Result)
    {
      //check for row events
      while (rc.MoveNext())
      {
        using var row = rc.Current;
        //etc
      }
    }
  }
  catch (TaskCanceledException)
  {
    //Handle cancellation as needed
  }
  cancel.Dispose();
}

{
  //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: 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();
  }
}