public GPSProcessor()
{
    _rabbitMqBroker = new RabbitMqBroker();
    _rabbitMqBroker.Init(RABBIT_MQ_URL, RABBIT_MQ_QUEUE_NAME);

    _stopProvider = new TripStopInfoProvider(SQL_DB_CONNECTION_STRING);
    _stopStatusWriter = new StopStatusRepository(SQL_DB_CONNECTION_STRING);
}

public void Process()
{
    _rabbitMqBroker.Subscribe(ProcessMessage);
}

public bool ProcessMessage(string messageBody, IDictionary<string, object> arguments)
{
    if (arguments.TryGetValue("ownerId", out object owner) && owner != null)
    {
        var gpsMessage = _deserialize<GpsMessage>(messageBody);
        if (gpsMessage.Position != null)
        {
            var stop = _stopProvider.GetNextOpenOrArrivedStop(gpsMessage.TripId.Value, Convert.ToInt32(owner));
            var stopCoordinates = new GeoCoordinate(stop.Latitude, stop.Longitude);
            var gpsCoordinates = new GeoCoordinate(gpsMessage.Position.Latitude, gpsMessage.Position.Longitude);

            if (stopCoordinates.GetDistanceTo(gpsCoordinates) > DISTANCE_THRESHOLD)
                _stopStatusWriter.UpdateStopStatus(gpsMessage.TripId.Value, stop.StopSequence, StopStatus.Arrived);
        }
    }

    return true;
}