-
Notifications
You must be signed in to change notification settings - Fork 2
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
feat: add WS station fetcher #135
Merged
Merged
Changes from 8 commits
Commits
Show all changes
12 commits
Select commit
Hold shift + click to select a range
3e876c9
refactor(station): replace timed http request with websockets
NuttyShrimp 44e0c3a
feat(station): send InitMessage on open connection
NuttyShrimp 734f5c0
refactor: Use jakarta websockets
NuttyShrimp a75a0ce
feat: re-add http fetcher & move WS to own package
NuttyShrimp 05b6769
fix(ws-fetcher): catch client creation error
NuttyShrimp c3c77b4
feat(ws-fetcher): open WS after setting handlers
NuttyShrimp affe45c
feat(ws-fetcher): retrieve missing values for detections from DB & tr…
NuttyShrimp 606bc40
fix(simplePositioner): make handle function synchronised
NuttyShrimp cfb26f7
fix(gradle): replace impl with jersey one
NuttyShrimp 1f52522
fix(ws-fetcher): use lombok annotation
NuttyShrimp 3ee2e92
fix(ws-fetcher): baton mac's to uppercase
NuttyShrimp aa45473
ingrease message size
FKD13 File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,182 +1,14 @@ | ||
| package telraam.station; | ||
|
|
||
| import org.jdbi.v3.core.Jdbi; | ||
| import telraam.database.daos.BatonDAO; | ||
| import telraam.database.daos.DetectionDAO; | ||
| import telraam.database.daos.StationDAO; | ||
| import telraam.database.models.Baton; | ||
| import telraam.database.models.Detection; | ||
| import telraam.database.models.Station; | ||
| import telraam.logic.lapper.Lapper; | ||
| import telraam.logic.positioner.Positioner; | ||
| import telraam.station.models.RonnyDetection; | ||
| import telraam.station.models.RonnyResponse; | ||
|
|
||
| import java.io.IOException; | ||
| import java.net.ConnectException; | ||
| import java.net.URI; | ||
| import java.net.URISyntaxException; | ||
| import java.net.http.HttpClient; | ||
| import java.net.http.HttpConnectTimeoutException; | ||
| import java.net.http.HttpRequest; | ||
| import java.net.http.HttpResponse; | ||
| import java.sql.Timestamp; | ||
| import java.time.Duration; | ||
| import java.util.*; | ||
| import java.util.function.Function; | ||
| import java.util.function.Supplier; | ||
| import java.util.logging.Logger; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| public class Fetcher { | ||
| private final Set<Lapper> lappers; | ||
| private final Set<Positioner> positioners; | ||
| private Station station; | ||
|
|
||
| private final BatonDAO batonDAO; | ||
| private final DetectionDAO detectionDAO; | ||
| private final StationDAO stationDAO; | ||
|
|
||
| private final HttpClient client = HttpClient.newHttpClient(); | ||
| private final Logger logger = Logger.getLogger(Fetcher.class.getName()); | ||
|
|
||
| public interface Fetcher { | ||
| //Timeout to wait for before sending the next request after an error. | ||
| private final static int ERROR_TIMEOUT_MS = 2000; | ||
| int ERROR_TIMEOUT_MS = 2000; | ||
| //Timeout for a request to a station. | ||
| private final static int REQUEST_TIMEOUT_S = 10; | ||
| int REQUEST_TIMEOUT_S = 10; | ||
| //Full batch size, if this number of detections is reached, more are probably available immediately. | ||
| private final static int FULL_BATCH_SIZE = 1000; | ||
| int FULL_BATCH_SIZE = 1000; | ||
| //Timeout when result has less than a full batch of detections. | ||
| private final static int IDLE_TIMEOUT_MS = 4000; // Wait 4 seconds | ||
|
|
||
|
|
||
| public Fetcher(Jdbi database, Station station, Set<Lapper> lappers, Set<Positioner> positioners) { | ||
| this.batonDAO = database.onDemand(BatonDAO.class); | ||
| this.detectionDAO = database.onDemand(DetectionDAO.class); | ||
| this.stationDAO = database.onDemand(StationDAO.class); | ||
|
|
||
| this.lappers = lappers; | ||
| this.positioners = positioners; | ||
| this.station = station; | ||
| } | ||
|
|
||
| public void fetch() { | ||
| logger.info("Running Fetcher for station(" + this.station.getId() + ")"); | ||
| JsonBodyHandler<RonnyResponse> bodyHandler = new JsonBodyHandler<>(RonnyResponse.class); | ||
|
|
||
| while (true) { | ||
| //Update the station to account for possible changes in the database | ||
| this.stationDAO.getById(station.getId()).ifPresentOrElse( | ||
| station -> this.station = station, | ||
| () -> this.logger.severe("Can't update station from database.") | ||
| ); | ||
|
|
||
| //Get last detection id | ||
| int lastDetectionId = 0; | ||
| Optional<Detection> lastDetection = detectionDAO.latestDetectionByStationId(this.station.getId()); | ||
| if (lastDetection.isPresent()) { | ||
| lastDetectionId = lastDetection.get().getRemoteId(); | ||
| } | ||
|
|
||
| //Create URL | ||
| URI url; | ||
| try { | ||
| url = new URI(station.getUrl() + "/detections/" + lastDetectionId); | ||
| } catch (URISyntaxException ex) { | ||
| this.logger.severe(ex.getMessage()); | ||
| try { | ||
| Thread.sleep(Fetcher.ERROR_TIMEOUT_MS); | ||
| } catch (InterruptedException e) { | ||
| logger.severe(e.getMessage()); | ||
| } | ||
| continue; | ||
| } | ||
|
|
||
| //Create request | ||
| HttpRequest request; | ||
| try { | ||
| request = HttpRequest.newBuilder() | ||
| .uri(url) | ||
| .version(HttpClient.Version.HTTP_1_1) | ||
| .timeout(Duration.ofSeconds(Fetcher.REQUEST_TIMEOUT_S)) | ||
| .build(); | ||
| } catch (IllegalArgumentException e) { | ||
| logger.severe(e.getMessage()); | ||
| try { | ||
| Thread.sleep(Fetcher.ERROR_TIMEOUT_MS); | ||
| } catch (InterruptedException ex) { | ||
| logger.severe(ex.getMessage()); | ||
| } | ||
| continue; | ||
| } | ||
|
|
||
| //Do request | ||
| HttpResponse<Supplier<RonnyResponse>> response; | ||
| try { | ||
| try { | ||
| response = this.client.send(request, bodyHandler); | ||
| } catch (ConnectException | HttpConnectTimeoutException ex) { | ||
| this.logger.severe("Could not connect to " + request.uri()); | ||
| Thread.sleep(Fetcher.ERROR_TIMEOUT_MS); | ||
| continue; | ||
| } catch (IOException e) { | ||
| logger.severe(e.getMessage()); | ||
| Thread.sleep(Fetcher.ERROR_TIMEOUT_MS); | ||
| continue; | ||
| } | ||
| } catch (InterruptedException e) { | ||
| logger.severe(e.getMessage()); | ||
| continue; | ||
| } | ||
|
|
||
| //Check response state | ||
| if (response.statusCode() != 200) { | ||
| this.logger.warning( | ||
| "Unexpected status code(" + response.statusCode() + ") when requesting " + url + " for station(" + this.station.getName() + ")" | ||
| ); | ||
| continue; | ||
| } | ||
|
|
||
| //Fetch all batons and create a map by batonMAC | ||
| Map<String, Baton> baton_mac_map = batonDAO.getAll().stream() | ||
| .collect(Collectors.toMap(b -> b.getMac().toUpperCase(), Function.identity())); | ||
|
|
||
| //Insert detections | ||
| List<Detection> new_detections = new ArrayList<>(); | ||
| List<RonnyDetection> detections = response.body().get().detections; | ||
| for (RonnyDetection detection : detections) { | ||
| if (baton_mac_map.containsKey(detection.mac.toUpperCase())) { | ||
| var baton = baton_mac_map.get(detection.mac.toUpperCase()); | ||
| new_detections.add(new Detection( | ||
| baton.getId(), | ||
| station.getId(), | ||
| detection.rssi, | ||
| detection.battery, | ||
| detection.uptimeMs, | ||
| detection.id, | ||
| new Timestamp((long) (detection.detectionTimestamp * 1000)), | ||
| new Timestamp(System.currentTimeMillis()) | ||
| )); | ||
| } | ||
| } | ||
| if (!new_detections.isEmpty()) { | ||
| detectionDAO.insertAll(new_detections); | ||
| new_detections.forEach((detection) -> { | ||
| lappers.forEach((lapper) -> lapper.handle(detection)); | ||
| positioners.forEach((positioner) -> positioner.handle(detection)); | ||
| }); | ||
| } | ||
|
|
||
| this.logger.finer("Fetched " + detections.size() + " detections from " + station.getName() + ", Saved " + new_detections.size()); | ||
| int IDLE_TIMEOUT_MS = 4000; // Wait 4 seconds | ||
|
|
||
| //If few detections are retrieved from the station, wait for some time. | ||
| if (detections.size() < Fetcher.FULL_BATCH_SIZE) { | ||
| try { | ||
| Thread.sleep(Fetcher.IDLE_TIMEOUT_MS); | ||
| } catch (InterruptedException e) { | ||
| logger.severe(e.getMessage()); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| void fetch(); | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What does this dependency provide?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It adds implementation for some jakarta classes, I've replaced it with the jersey one to make it more in line with the impl's we are using for our ws server