Source Worker and Report Generator

A source worker manages the connection to an event source and fetches events from it. An event is a direct representation of the original information published by the event source, and it has to be converted into a report before it can be displayed in CysTerra. This conversion is done by a report generator.

The source workers and the report generators for the same event source are usually implemented in the same extension.

Source Worker

A source worker is a class implementing the ISourceWorker interface. The following is an example implementation.

using Cryville.EEW;
using System;
using System.Diagnostics.CodeAnalysis;
using System.Globalization;
using System.Threading;
using System.Threading.Tasks;

namespace MyExtension {
	public class MyWorker : ISourceWorker {
		public string? GetName([NotNull] ref CultureInfo? culture) {
			// Get the name of the source worker from the resources
			using var lres = new LocalizedResource("", ref culture);
			var res = lres.RootMessageStringSet;
			return res.GetStringRequired("SourceName");
		}

		public event Handler<object?>? Received;
		public event Handler<Heartbeat>? Heartbeat;
		public event Handler<Exception>? ErrorEmitted;

		public async Task RunAsync(CancellationToken cancellationToken) {
			// Signal that the worker is connected
			Heartbeat?.Invoke(this, new());
			// Emit an object as an event
			Received?.Invoke(this, new object());
		}
	}
}

When this worker is started, it emits a new object() as an event, and then exits.

Normally a source worker does something much more complicated, usually in a loop to fetch events periodically.

public async Task RunAsync(CancellationToken cancellationToken) {
	try {
		while (true) {
			// ...
			// Fetch and parse new events if there is any
			// ...
			
			// Wait before next request
			await Task.Delay(TimeSpan.FromSeconds(60), cancellationToken).ConfigureAwait(true);
		}
	}
	catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) {
		// Do nothing: Worker task cancellation requested
	}
}

Caution

Source workers are NOT designed to fetch continuous real-time data, such as real-time intensity of seismic stations. Doing this can lead to very bad performance.

A new component will be implemented to fetch these real-time data in the future.

A source worker is built with a builder exported with [Export(typeof(IBuilder<ISourceWorker>))].

Built-in Base Workers

As many event sources publish events in HTTP or WebSocket, CysTerra have built in two classes to fetch events from these protocols respectively for convenience. Inherit your source worker from HttpPullWorker or WebSocketWorker to make use of them.

HttpPullWorker
WebSocketWorker

HttpPullWorker fetches events from the given URI by sending GET requests periodically, and then passes the response to the Handle method if its status code is 200 (OK).

using Cryville.EEW;
using System;
using System.Diagnostics.CodeAnalysis;
using System.Globalization;
using System.IO;
using System.Net;
using System.Net.Http;
using System.Net.Http.Headers;
using System.Threading;
using System.Threading.Tasks;

namespace MyExtension {
	public class MyHttpWorker(Uri uri) : HttpPullWorker(uri), ISourceWorker {
		public string? GetName([NotNull] ref CultureInfo? culture) {
			// Get the name of the source worker from the resources
			using var lres = new LocalizedResource("", ref culture);
			var res = lres.RootMessageStringSet;
			return res.GetStringRequired("SourceName");
		}

		public event Handler<object?>? Received;
		public event Handler<Heartbeat>? Heartbeat;
		public event Handler<Exception>? ErrorEmitted;

		protected override void OnHeartbeat() => Heartbeat?.Invoke(this, new());
		protected override void OnError(Exception ex) => ErrorEmitted?.Invoke(this, ex);

		protected override async Task Handle(Stream stream, HttpResponseHeaders headers, CancellationToken cancellationToken) {
			// ...
			// Deserialize the event from the response stream
			// and raise the event with Received?.Invoke()
			// ...
		}
	}
}

By default, HttpPullWorker inspects the max-age directive in the Cache-Control response header to determine the period of requesting, and falls back to 60 seconds if the directive is not found. You can change this behavior by overriding the ForceDefaultPeriod, DefaultPeriod, and MinimumPeriod properties.

If you want to handle requests with status codes other than 200 (OK), override the HandleRawResponse method. Call the base method at the end of your override.

protected override Task HandleRawResponse(HttpResponseMessage response, CancellationToken cancellationToken) {
	ThrowHelper.ThrowIfNull(response);
	if (response.StatusCode == HttpStatusCode.Unauthorized) {
		throw new SourceWorkerClientException("Authorization failed.");
	}
	return base.HandleRawResponse(response, cancellationToken);
}

You can modify the URI to be requested by HttpPullWorker dynamically by overriding the GetUri method.

You can send additional requests by calling the TryGetAsync method or the TrySendAsync method.

WebSocketWorker receives messages from the given URI, and passes each message to the Handle method.

using Cryville.EEW;
using System;
using System.Diagnostics.CodeAnalysis;
using System.Globalization;
using System.IO;
using System.Threading;
using System.Threading.Tasks;

namespace MyExtension {
	public class MyWebSocketWorker(Uri uri) : WebSocketWorker(uri), ISourceWorker {
		public string? GetName([NotNull] ref CultureInfo? culture) {
			// Get the name of the source worker from the resources
			using var lres = new LocalizedResource("", ref culture);
			var res = lres.RootMessageStringSet;
			return res.GetStringRequired("SourceName");
		}

		public event Handler<object?>? Received;
		public event Handler<Heartbeat>? Heartbeat;
		public event Handler<Exception>? ErrorEmitted;

		protected override void OnHeartbeat() => Heartbeat?.Invoke(this, new());
		protected override void OnError(Exception ex) => ErrorEmitted?.Invoke(this, ex);

		protected override async Task Handle(Stream stream, CancellationToken cancellationToken) {
			// ...
			// Deserialize the event from the response stream
			// and raise the event with Received?.Invoke()
			// ...
		}
	}
}

Errors

Raising ErrorEmitted with an instance of SourceWorkerNetworkException indicates a non-fatal network error (this error is not reportable), and the source worker is considered disconnected with the event source, until Heartbeat is raised afterwards, which indicates a successful reconnection. Raising ErrorEmitted with any other exceptions indicates an non-fatal non-network error.

A faulted RunAsync task (i.e. the task exits with an unhandled exception) indicates a fatal error. If it is faulted with an instance of SourceWorkerClientException, the error is considered to be caused by the user and is not reportable.

Note

For reportable and non-reportable errors, see Error Reporting.

Report Generator

A report generator is a class implementing the IGenerator<ReportModel> interface or the IContextedGenerator<IReportGeneratorContext, ReportModel> interface. The following is an example implementation.

using Cryville.Common.Compat;
using Cryville.EEW;
using Cryville.EEW.Report;
using System;
using System.Globalization;

namespace MyExtension {
	public class MyReportGenerator : IContextedGenerator<MyEvent, IReportGeneratorContext, ReportModel> {
		public ReportModel Generate(MyEvent e, IReportGeneratorContext? context, ref CultureInfo culture) {
			ThrowHelper.ThrowIfNull(e);
			context ??= EmptyReportGeneratorContext.Instance;

			using var lres = new LocalizedResource("", ref culture);
			var res = lres.RootMessageStringSet;
			var result = new ReportModel {
				Title = res.GetStringRequired("Title"),
				Source = res.GetStringRequired("AuthorityName"),
				Location = /* ... */,
				Time = /* ... */,
				TimeZone = /* ... */,
			};
			result.GroupKeys.Add(/* ... */);
			result.Properties.Add(/* ... */);

			return result;
		}
	}
}

A report is displayed in the report list like the following.

Title | Source

Key Prop

1.0

Condition

Location Predicate

2000-01-01 00:00:00 (UTC)

Prop1 2.0 Prop2 3.0

“Title”: Title
“Source”: Source
“#1”: RevisionKey
“Location”: Location
“Predicate”: Predicate
“2000-01-01 00:00:00 (UTC)”: Time
“Key Prop 1.0 Condition”: Properties[0]
“Prop1 2.0”: Properties[1]
“Prop2 3.0”: Properties[2]

For more information, see the API documentation of the ReportModel class.

A report generator is built with a builder exported with [Export(typeof(IBuilder<IGenerator<ReportModel>>))].

Report Grouping

Related reports are grouped together for easier browsing. A report revising another report is grouped into the same report unit with that report, and relevant report units are grouped into a report group.

Report grouping is based on the GroupKeys property in reports. Two reports with any matching group keys (IReportGroupKey) are grouped into the same report group.

IReportUnitKey, derived from IReportGroupKey, is for grouping reports into report units. Two reports with the same unit key are grouped into the same report unit.

Report Validity

Invalidated reports are collapsed in CysTerra. A report is considered invalidated if either:

Each of its unit key is covered by any unit key in another report. (IsCoveredBy)
InvalidatedTime is set and the time now is later than it.

If InvalidatedTime is set and the time now is earlier than it, the report is pinned as an ongoing event and is always displayed on the map despite whether it is selected or not, until it is invalidated.

Table of Contents