Exception when Internet Connection Disconnected

[MohMehrnia]

Hi, I use the below code for stream data and if the internet connection is broken, the function throws an exception and the streaming process stoped.
how do prevent this procedure?

        public async static void SyncData(string startDate, string endDate, CancellationToken cancellationToken = default)
        {
            try
            {
                using (var store = new DocumentStore
                {
                    Urls = new[] { ConnectionsSetting.RavenDBServer },
                    Database = ConnectionsSetting.wellName,
                    Certificate = cert,
                    Conventions =
                {
                    UseCompression=true
                }
                })
                {
                    store.Initialize();
                    using (IAsyncDocumentSession Session = store.OpenAsyncSession(new SessionOptions
                    {
                        NoCaching = true
                    }))
                    {
                        LocalDatabaseManagement.AllowWriteData = true;

                        IAsyncRawDocumentQuery<WellRecord> query = Session.Advanced.AsyncRawQuery<WellRecord>($"from WellRecords where Rdtime between '{startDate}' and '{endDate}' order by Rdtime desc ");
                        IAsyncEnumerator<StreamResult<WellRecord>> results = await Session.Advanced.StreamAsync(query, cancellationToken);
                        while (await results.MoveNextAsync())
                        {
                            if (cancellationToken.IsCancellationRequested)
                                break;
                            StreamResult<WellRecord> wellRecord = results.Current;
                            HandleSyncRecord?.Invoke(null, new CosumeEventArg { WellRecord = wellRecord.Document });
                        }
                        HandleSyncRecord?.Invoke(null, new CosumeEventArg { WellRecord = null });
                    };
                }
            }
            catch { }
        }

[ayende]

I'm not sure what do you expect RavenDB to do at this point. You are trying to stream all the records, but you have unreliable connection. The Streaming API is a single continuous operation, any network outage will cause it it be broken.

Note that RavenDB offers an API that is meant to handle such disconnect scenario and retry from the last acknowledged location. That is what the Subscription API is for.

Would be interesting to understand more about your scenario.

[MohMehrnia]

When I used subscription API, I can simulate stream read data (i have a time series collection and I need stream data between two dates, now subscription help me ?)

[ayende]

I don't understand the question. What is the exact query you are trying to do?

[MohMehrnia]

IAsyncRawDocumentQuery<WellRecord> query = Session.Advanced.AsyncRawQuery<WellRecord>($"from WellRecords where Rdtime between '{startDate}' and '{endDate}' order by Rdtime desc ");
I need to stream old data (like the above query) but with network failed procedure or reset connection or reset procedure,

[ayende]

That isn't a typical subscription query. You can make it work, but it won't be nice. Subscriptions also don't support order by.

I think we need to go a step back. What is the scenario where you need to recover from network interruption?

Also, you can simple remember the last date of the document you received and retry the query from that point.

Finally, please use $startDate / $endDate parameters and not string conact.

[MohMehrnia]

when internet is disconnected, stream process throw exception and when net is back stream can not be continue

[ayende]

The stream is basically a TCP connection, it isn't expected to handle network disruptions like that.

[MohMehrnia]

i need pause streaming when internet disconnected and when internet is back streaming resume again

[ayende]

Not something that is going to work. You can remember the last date of the document that you got and issue another query from there.

[MohMehrnia]

when throwing exception, my app can not continue streaming even run function again and for continue must be reset
app

[MohMehrnia]

public async static void SyncData(string startDate, string endDate, CancellationToken cancellationToken = default)
        {
            DateTime LastSyncDatetime = DateTime.Parse(startDate);
            try
            {
                using (var store = new DocumentStore
                {
                    Urls = new[] { ConnectionsSetting.RavenDBServer },
                    Database = ConnectionsSetting.wellName,
                    Certificate = cert,
                    Conventions =
                {
                    UseCompression=true
                }
                })
                {
                    store.Initialize();
                    using (IAsyncDocumentSession Session = store.OpenAsyncSession(new SessionOptions
                    {
                        NoCaching = true
                    }))
                    {
                        LocalDatabaseManagement.AllowWriteData = true;

                        IAsyncRawDocumentQuery<WellRecord> query = Session.Advanced.AsyncRawQuery<WellRecord>($"from WellRecords where Rdtime between '{startDate}' and '{endDate}' order by Rdtime desc ");
                        IAsyncEnumerator<StreamResult<WellRecord>> results = await Session.Advanced.StreamAsync(query, cancellationToken);
                        while (await results.MoveNextAsync())
                        {
                            if (cancellationToken.IsCancellationRequested)
                                break;
                            StreamResult<WellRecord> wellRecord = results.Current;
                            LastSyncDatetime = wellRecord.Document.Rdtime;
                            HandleSyncRecord?.Invoke(null, new CosumeEventArg { WellRecord = wellRecord.Document });
                        }
                        HandleSyncRecord?.Invoke(null, new CosumeEventArg { WellRecord = null });
                    };
                }
            }
            catch 
            {
                SyncData(LastSyncDatetime.ToString("s"), endDate, cancellationToken);
            }

[MohMehrnia]

this my function, and when catch run function from last record started, but streaming not started

[ayende]

You need to handle that scenario, and retry the query from where you last stopped.

Also, please reply to the previous comment, don't start new ones.

[MohMehrnia]

please, more explain

[ayende]

You can do something like this:

public async static void SyncData(string startDate, string endDate, CancellationToken cancellationToken = default)
{
    DateTime LastSyncDatetime = DateTime.Parse(startDate);

    using (var store = new DocumentStore
    {
        Urls = new[] { ConnectionsSetting.RavenDBServer },
        Database = ConnectionsSetting.wellName,
        Certificate = cert,
        Conventions =
    {
        UseCompression=true
    }
    })
    {
        store.Initialize();
        while(true)
        {
            using (IAsyncDocumentSession Session = store.OpenAsyncSession(new SessionOptions
            {
                NoCaching = true
            }))
            {
                LocalDatabaseManagement.AllowWriteData = true;

                IAsyncRawDocumentQuery<WellRecord> query = Session.Advanced.AsyncRawQuery<WellRecord>(
                    $"from WellRecords where Rdtime between $start and $end order by Rdtime desc ")
                    .AddParameter("start", startDate)
                    .AddParameter("end", endDate);
                IAsyncEnumerator<StreamResult<WellRecord>> results;
                try
                {
                    results = await Session.Advanced.StreamAsync(query, cancellationToken);
                }
                catch(Exception e)
                {
                    await Task.Delay(100);
                    HandleError?.Invoke(e);
                    continue; // retry... 
                }
                while (true)
                {
                    try
                    {
                        if(await results.MoveNextAsync() == false)
                            return;
                    }
                    catch(Exception e)
                    {
                        await Task.Delay(100);
                        HandleError?.Invoke(e);
                        break; // retry... 
                    }
                    
                    if (cancellationToken.IsCancellationRequested)
                        return;
                    StreamResult<WellRecord> wellRecord = results.Current;
                    startDate = wellRecord.Document.Rdtime;
                    HandleSyncRecord?.Invoke(null, new CosumeEventArg { WellRecord = wellRecord.Document });
                }
                HandleSyncRecord?.Invoke(null, new CosumeEventArg { WellRecord = null });
            };
        }
    }
}