6048833209

Committed 01 Sep 2023 11:23AM UTC coverage: 73.892% (-20.7%) from 94.619%

Build # 6048833209

Build Type

push

github

Committed by

davewalker5

Commit Message

Added aircraft altitude and position recording

Run Details

169 of 192 branches covered (0.0%)

Branch coverage included in aggregate %.

271 of 271 new or added lines in 8 files covered. (100.0%)

581 of 823 relevant lines covered (70.6%)

14.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.73

/src/BaseStationReader.Logic/QueuedWriter.cs

﻿using BaseStationReader.Entities.Events;
using BaseStationReader.Entities.Interfaces;
using BaseStationReader.Entities.Logging;
using BaseStationReader.Entities.Tracking;
using System.Collections.Concurrent;
using System.Diagnostics;

namespace BaseStationReader.Logic
{
    public class QueuedWriter : IQueuedWriter
    {
        private readonly IAircraftWriter _aircraftWriter;
        private readonly IPositionWriter _positionWriter;
        private readonly ConcurrentQueue<object> _queue = new ConcurrentQueue<object>();
        private readonly ITrackerLogger _logger;
        private readonly ITrackerTimer _timer;
        private readonly int _batchSize = 0;

        public event EventHandler<BatchWrittenEventArgs>? BatchWritten;

        public QueuedWriter(
            IAircraftWriter aircraftWriter,
            IPositionWriter positionWriter,
            ITrackerLogger logger,
            ITrackerTimer timer,
            int batchSize)
        {
            _aircraftWriter = aircraftWriter;
            _positionWriter = positionWriter;
            _logger = logger;
            _timer = timer;
            _timer.Tick += OnTimer;
            _batchSize = batchSize;
        }

        /// <summary>
        /// Put tracking details into the queue to be written
        /// </summary>
        /// <param name="aircraft"></param>
        public void Push(object entity)
        {
            // To stop the queue growing and consuming memory, entries are discarded if the timer
            // hasn't been started. Also, check the object being pushed is a valid tracking entity
            if ((_timer != null) && ((entity is Aircraft) || (entity is AircraftPosition)))
            {
                _queue.Enqueue(entity);
            }
        }

        /// <summary>
        /// 
        /// </summary>
        public async void Start()
        {
            // Set the locked flag on all unlocked records. This prevents confusing tracking of the same aircraft
            // on different flights
            List<Aircraft> unlocked = await _aircraftWriter.ListAsync(x => !x.Locked);
            foreach (var aircraft in unlocked)
            {
                aircraft.Locked = true;
                _queue.Enqueue(aircraft);
            }

            // Now start the timer
            _timer.Start();
        }

        /// <summary>
        /// Stop writing to the tracking database
        /// </summary>
        public void Stop()
        {
            _timer.Stop();
            _queue.Clear();
        }

        /// <summary>
        /// When the timer fires, write the next batch of records to the database
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnTimer(object? sender, EventArgs e)
        {
            _timer.Stop();

            // Time how long the update takes
            Stopwatch stopwatch = Stopwatch.StartNew();

            // Iterate over at most the current batch size entries in the queue
            var initialQueueSize = _queue.Count;
            for (int i = 0; i < _batchSize; i++)
            {
                // Attempt to get the next item and if it's not there break out
                if (!_queue.TryDequeue(out object? queued))
                {
                    break;
                }

                // Write the dequeued object to the database
                Task.Run(() => WriteDequeuedObject(queued)).Wait();
            }
            stopwatch.Stop();
            var finalQueueSize = _queue.Count;

            try
            {
                // Notify subscribers that a batch has been written
                BatchWritten?.Invoke(this, new BatchWrittenEventArgs
                {
                    InitialQueueSize = initialQueueSize,
                    FinalQueueSize = finalQueueSize,
                    Duration = stopwatch.ElapsedMilliseconds
                });
            }
            catch (Exception ex)
            {
                // Log and sink the exception. The writer has to be protected from errors in the
                // subscriber callbacks or the application will stop updating
                _logger.LogException(ex);
            }

            _timer.Start();
        }

        /// <summary>
        /// Receive a de-queued object, determine its type and use the appropriate writer to write it
        /// </summary>
        /// <param name="queued"></param>
        private async void WriteDequeuedObject(object queued)
        {
            // If it's an aircraft and it's an existing record that hasn't been locked, get the ID for update
            Aircraft? aircraft = queued as Aircraft;
            AircraftPosition? position = null;
            if (aircraft != null)
            {
                // See if this is an existing aircraft for which the record hasn't been locked, to get the ID for update
                aircraft.Id = await MatchAircraftAddress(aircraft.Address, false);
            }
            else
            {
                // Not an aircraft so it must be a position. If the aircraft Id isn't specified, try to find a match
                // for the address. Note that we allow matches to locked aircraft as the final position update may
                // come after the aircraft record has been locked
                position = queued as AircraftPosition;
                if (position != null)
                {
                    position.AircraftId = await MatchAircraftAddress(position.Address, true);
                }
            }

            // Write the data to the database
            try
            {
                if (aircraft != null)
                {
                    _logger.LogMessage(Severity.Debug, $"Writing aircraft {aircraft.Address} with Id {aircraft.Id}");
                    await _aircraftWriter.WriteAsync(aircraft);
                }
                else if ((position != null) && (position.AircraftId > 0))
                {
                    _logger.LogMessage(Severity.Debug, $"Writing position for aircraft with Id {position.AircraftId}");
                    await _positionWriter.WriteAsync(position);
                }
            }
            catch (Exception ex)
            {
                // Log and sink the exception. The writer needs to continue or the application will
                // stop writing to the database
                _logger.LogException(ex);
            }
        }

        /// <summary>
        /// Find an existing aircraft with the specified address and return it's Id, if found, or 0 if not found
        /// </summary>
        /// <param name="address"></param>
        /// <param name="matchLockedAircraft"></param>
        /// <returns></returns>
        private async Task<int> MatchAircraftAddress(string address, bool matchLockedAircraft)
        {
            int matchingId = 0;

            if (!string.IsNullOrEmpty(address))
            {
                var existing = await _aircraftWriter.GetAsync(x => (x.Address == address) && (matchLockedAircraft || !x.Locked));
                if (existing != null)
                {
                    matchingId = existing.Id;
                }
            }

            return matchingId;
        }
    }
}

1	using BaseStationReader.Entities.Events;
2	using BaseStationReader.Entities.Interfaces;
3	using BaseStationReader.Entities.Logging;
4	using BaseStationReader.Entities.Tracking;
5	using System.Collections.Concurrent;
6	using System.Diagnostics;
7
8	namespace BaseStationReader.Logic
9	{
10	public class QueuedWriter : IQueuedWriter
11	{
12	private readonly IAircraftWriter _aircraftWriter;
13	private readonly IPositionWriter _positionWriter;
14	private readonly ConcurrentQueue<object> _queue = new ConcurrentQueue<object>();	2✔
15	private readonly ITrackerLogger _logger;
16	private readonly ITrackerTimer _timer;
17	private readonly int _batchSize = 0;	2✔
18
19	public event EventHandler<BatchWrittenEventArgs>? BatchWritten;
20
21	public QueuedWriter(	2✔
22	IAircraftWriter aircraftWriter,	2✔
23	IPositionWriter positionWriter,	2✔
24	ITrackerLogger logger,	2✔
25	ITrackerTimer timer,	2✔
26	int batchSize)	2✔
27	{	2✔
28	_aircraftWriter = aircraftWriter;	2✔
29	_positionWriter = positionWriter;	2✔
30	_logger = logger;	2✔
31	_timer = timer;	2✔
32	_timer.Tick += OnTimer;	2✔
33	_batchSize = batchSize;	2✔
34	}	2✔
35
36	/// <summary>
37	/// Put tracking details into the queue to be written
38	/// </summary>
39	/// <param name="aircraft"></param>
40	public void Push(object entity)
41	{	4✔
42	// To stop the queue growing and consuming memory, entries are discarded if the timer
43	// hasn't been started. Also, check the object being pushed is a valid tracking entity
44	if ((_timer != null) && ((entity is Aircraft) \|\| (entity is AircraftPosition)))	4!
45	{	4✔
46	_queue.Enqueue(entity);	4✔
47	}	4✔
48	}	4✔
49
50	/// <summary>
51	///
52	/// </summary>
53	public async void Start()
54	{	3✔
55	// Set the locked flag on all unlocked records. This prevents confusing tracking of the same aircraft
56	// on different flights
57	List<Aircraft> unlocked = await _aircraftWriter.ListAsync(x => !x.Locked);	3✔
58	foreach (var aircraft in unlocked)	11✔
59	{	1✔
60	aircraft.Locked = true;	1✔
61	_queue.Enqueue(aircraft);	1✔
62	}	1✔
63
64	// Now start the timer
65	_timer.Start();	3✔
66	}	3✔
67
68	/// <summary>
69	/// Stop writing to the tracking database
70	/// </summary>
71	public void Stop()
72	{	3✔
73	_timer.Stop();	3✔
74	_queue.Clear();	3✔
75	}	3✔
76
77	/// <summary>
78	/// When the timer fires, write the next batch of records to the database
79	/// </summary>
80	/// <param name="sender"></param>
81	/// <param name="e"></param>
82	private void OnTimer(object? sender, EventArgs e)
83	{	5✔
84	_timer.Stop();	5✔
85
86	// Time how long the update takes
87	Stopwatch stopwatch = Stopwatch.StartNew();	5✔
88
89	// Iterate over at most the current batch size entries in the queue
90	var initialQueueSize = _queue.Count;	5✔
91	for (int i = 0; i < _batchSize; i++)	18✔
92	{	9✔
93	// Attempt to get the next item and if it's not there break out
94	if (!_queue.TryDequeue(out object? queued))	9✔
95	{	5✔
96	break;	5✔
97	}
98
99	// Write the dequeued object to the database
100	Task.Run(() => WriteDequeuedObject(queued)).Wait();	8✔
101	}	4✔
102	stopwatch.Stop();	5✔
103	var finalQueueSize = _queue.Count;	5✔
104
105	try
106	{	5✔
107	// Notify subscribers that a batch has been written
108	BatchWritten?.Invoke(this, new BatchWrittenEventArgs	5!
109	{	5✔
110	InitialQueueSize = initialQueueSize,	5✔
111	FinalQueueSize = finalQueueSize,	5✔
112	Duration = stopwatch.ElapsedMilliseconds	5✔
113	});	5✔
114	}	5✔
115	catch (Exception ex)	×
116	{	×
117	// Log and sink the exception. The writer has to be protected from errors in the
118	// subscriber callbacks or the application will stop updating
119	_logger.LogException(ex);	×
120	}	×
121
122	_timer.Start();	5✔
123	}	5✔
124
125	/// <summary>
126	/// Receive a de-queued object, determine its type and use the appropriate writer to write it
127	/// </summary>
128	/// <param name="queued"></param>
129	private async void WriteDequeuedObject(object queued)
130	{	4✔
131	// If it's an aircraft and it's an existing record that hasn't been locked, get the ID for update
132	Aircraft? aircraft = queued as Aircraft;	4✔
133	AircraftPosition? position = null;	4✔
134	if (aircraft != null)	4✔
135	{	3✔
136	// See if this is an existing aircraft for which the record hasn't been locked, to get the ID for update
137	aircraft.Id = await MatchAircraftAddress(aircraft.Address, false);	3✔
138	}	3✔
139	else
140	{	1✔
141	// Not an aircraft so it must be a position. If the aircraft Id isn't specified, try to find a match
142	// for the address. Note that we allow matches to locked aircraft as the final position update may
143	// come after the aircraft record has been locked
144	position = queued as AircraftPosition;	1✔
145	if (position != null)	1✔
146	{	1✔
147	position.AircraftId = await MatchAircraftAddress(position.Address, true);	1✔
148	}	1✔
149	}	1✔
150
151	// Write the data to the database
152	try
153	{	4✔
154	if (aircraft != null)	4✔
155	{	3✔
156	_logger.LogMessage(Severity.Debug, $"Writing aircraft {aircraft.Address} with Id {aircraft.Id}");	3✔
157	await _aircraftWriter.WriteAsync(aircraft);	3✔
158	}	3✔
159	else if ((position != null) && (position.AircraftId > 0))	1!
160	{	1✔
161	_logger.LogMessage(Severity.Debug, $"Writing position for aircraft with Id {position.AircraftId}");	1✔
162	await _positionWriter.WriteAsync(position);	1✔
163	}	1✔
164	}	4✔
165	catch (Exception ex)	×
166	{	×
167	// Log and sink the exception. The writer needs to continue or the application will
168	// stop writing to the database
169	_logger.LogException(ex);	×
170	}	×
171	}	4✔
172
173	/// <summary>
174	/// Find an existing aircraft with the specified address and return it's Id, if found, or 0 if not found
175	/// </summary>
176	/// <param name="address"></param>
177	/// <param name="matchLockedAircraft"></param>
178	/// <returns></returns>
179	private async Task<int> MatchAircraftAddress(string address, bool matchLockedAircraft)
180	{	4✔
181	int matchingId = 0;	4✔
182
183	if (!string.IsNullOrEmpty(address))	4✔
184	{	4✔
185	var existing = await _aircraftWriter.GetAsync(x => (x.Address == address) && (matchLockedAircraft \|\| !x.Locked));	4✔
186	if (existing != null)	4✔
187	{	2✔
188	matchingId = existing.Id;	2✔
189	}	2✔
190	}	4✔
191
192	return matchingId;	4✔
193	}	4✔
194	}
195	}

davewalker5 / ADS-B-BaseStationReader / 6048833209

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous