prepare release 1.3.3

Plus some minor test dependency updates.

Cherry-picked from 55271011b6

.drone.yml

@@ -0,0 +1,31 @@
+kind: pipeline
+type: docker
+name: java8
+
+steps:
+  - name: test
+    image: maven:3-eclipse-temurin-8
+    commands:
+      - mvn -B clean test
+
+---
+kind: pipeline
+type: docker
+name: java11
+
+steps:
+  - name: test
+    image: maven:3-eclipse-temurin-11
+    commands:
+      - mvn -B clean test
+
+---
+kind: pipeline
+type: docker
+name: java17
+
+steps:
+  - name: test
+    image: maven:3-eclipse-temurin-17
+    commands:
+      - mvn -B clean test

.github/workflows/ci.yml

@@ -0,0 +1,33 @@
+name: CI
+on: [ push, pull_request ]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        jdk: [ 8, 11, 17 ]
+        include:
+          - jdk: 11
+            analysis: true
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+      - name: Set up Java
+        uses: actions/setup-java@v3
+        with:
+          java-version: ${{ matrix.jdk }}
+          distribution: 'temurin'
+      - name: Test
+        run: mvn -B -P coverage clean verify
+      - name: Analysis
+        if: matrix.analysis
+        run: >
+          mvn -B sonar:sonar
+          -Dsonar.host.url=https://sonarcloud.io
+          -Dsonar.organization=stklcode-github
+          -Dsonar.login=$SONAR_TOKEN
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}

.travis.yml

@@ -1,15 +0,0 @@
-language: java
-jdk:
-  - oraclejdk8
-install: true
-addons:
-  sonarcloud:
-    organization: "stklcode-github"
-    token:
-        secure: "FkEe/+MKpF4pSX3ZYOgu7oeIKf0460Q3XVLUhIX9bk2dyY8hoab74oCo4FtD7jim0+ZC13JVHGDX7iOQMUtS5EZ+x+pA0qpppzCK5zV8afN/l46HJ07kJldvr+EH0klbDVMFZQ5dT7r/w6CoDzjtENHzKQAJLcheUVDNpkcuBdaplTqIAVf3lQpKtOuVjQJ5qZDwwS5wsHNqPcYbcEGrPmcKDVnp3mD3bfI6dT1bbRt845QcD73rPYkQKxen8eIwJxFf5MZStgvbj7yphPxPGwoLAsoLP6LpThTDYcrg+vgUnSs+l9GckL3IbhPAmecixLbKVnphBZzRTvpdMTt5KeOoAJ2nM6RLs5cRCqiEgLEioWkVaSH5WxoBj38Z1h4fTsDV3dTcCuQWX8GFxdeeTelu+XbatdRWMnUgiF7oax+uNvR62fasTbAc7dWPJbARiD7ZbkWH4nHEY07xKKx87xzUz36ZeEHGoBXgqnLmv/FCwqMrEpOoIT41fc0WYtdIA4wjRoAyG0u+wNBMbVlf4PK72seM4b/bmU+TtmaaVla/SvNOiz+A3DHxtJEUScPcL3QGjviddglMf+wyD6VXVViq9VuYRKZFyjpuoNpb5lwEbwmRnmLabBx8jBgyPinjpmqHYlIntcPAwuyLRaqTHFcmCrbeeZEf7KaPRYKx+Cs="
-cache:
-  directories:
-    - '$HOME/.m2/repository'
-    - '$HOME/.sonar/cache'
-script:
-  - mvn -P jacoco clean package sonar:sonar

CHANGELOG.md

@@ -1,12 +1,80 @@
-## 1.1.1 [2018-02-20]
-* [improvement] On connection or parsing errors, the `IOException` is no longer ignored, but encapsulated in `RuntimeException` (no StackTraces printed)
-* [internal] Code cleanup and minor improvements
-* [dependeny] Minor dependency updates
+# Changelog
+All notable changes to this project will be documented in this file.
 
-## 1.1.0 [2017-01-07]
-* [feature] Filter stops by coordinates and radius
-* [feature] Filter trips by destination and and towards fields
-* [test] Test coverage 100% (line); tested against ASEAG and TFL APIs
+## 1.3.3 - 2022-11-21
+### Security
+* Updated Jackson dependency to 2.14.0
 
-## 1.0.0 [2017-01-02]
-* Initial release
+### Fixed
+* Querying trips and messages with limit directly from `Query` instance (#19)
+
+
+## 1.3.2 - 2022-08-30
+
+### Improvements
+* Dependency updates
+
+
+## 1.3.1 - 2020-12-12
+### Fixed
+* Allow reopening an `AsyncUraTripReader` without raising an exception (#13)
+
+### Improvements
+* Dependency updates
+
+
+## 1.3.0 - 2019-12-04
+### Security
+* Updated dependencies
+
+### Features
+* Added support for reading messages, using `getMessages()` method (#5)
+
+
+## 1.2.0 - 2019-06-20
+### Security
+* Updated dependencies
+
+### Features
+* Added support for stream API with asynchronous reader, using `getTripsStream()` method (#1)
+
+
+## 1.1.4 - 2018-11-19
+### Fixed
+* Fixed issue with direction ID as `String` instead if `Integer` (#2)
+* Fixed issue with vehicle ID being `null` (#3)
+* Fixed issue with spaces in search parameters (#4)
+
+
+## 1.1.3 - 2018-11-13
+### Security
+* Updates Jackson dependency 2.9.4 to 2.9.7 (CVE-2018-7489)
+
+### Improvements
+* Client and model classes implement `Serializable`
+* Dependency updates
+
+
+## 1.1.2 - 2018-03-24
+### Improvements
+* Added automatic module name for JPMS compatibility
+
+
+## 1.1.1 - 2018-02-20
+### Improvements
+* On connection or parsing errors, the `IOException` is no longer ignored, but encapsulated in `RuntimeException` (no StackTraces printed)
+* Code cleanup and minor improvements
+* Minor dependency updates
+
+
+## 1.1.0 - 2017-01-07
+### Features
+* Filter stops by coordinates and radius
+* Filter trips by destination and and towards fields
+
+### Misc
+* Test coverage 100% (line); tested against ASEAG and TFL APIs
+
+
+## 1.0.0 - 2017-01-02
+* Initial release

CONTRIBUTING.md

@@ -0,0 +1,118 @@
+# How to contribute
+
+As for all great Open Source projects, contributions in form of bug reports and code are welcome and important to keep the project alive.
+
+In general, this project follows the [GitHub Flow](https://guides.github.com/introduction/flow/). 
+Fork the project, commit your changes to your branch, open a pull request and it will probably be merged.
+However, to ensure maintainability and quality of the code, there are some guidelines you might be more or less familiar with. 
+For that purpose, this document describes the important points.
+
+
+## Opening an Issue
+
+If you experience any issues with the plugin or the code, don't hesitate to file an issue.
+
+### Bug Reports
+
+Think you found a bug?
+Please clearly state what happens and describe your environment to help tracking down the issue.
+ 
+* Which version of the project are you running?
+* Which version of Java?
+* Which API do you try to query?
+
+### Feature Requests
+
+Missing a feature or like to have certain functionality enhanced?
+No problem, please open an issue and describe what and why you think this change is required.
+
+
+## Pull Requests
+
+If you want to contribute your code to solve an issue or implement a desired feature yourself, you might open a pull request.
+If the changes introduce new functionality or affect major parts of existing code, please consider opening an issue for discussion first.
+
+For adding new functionality a new test case the corresponding JUnit test would be nice (no hard criterion though).
+
+### Branches
+
+The `master` branch represents the current state of development.
+Please ensure your initial code is up to date with it at the time you start development.
+The `master` should also be target for most pull requests.
+
+In addition, this project features a `develop` branch, which holds bleeding edge developments, not necessarily considered stable or even compatible.
+Do not expect this code to run smoothly, but you might have a look into the history to see if some work on an issue has already been started there.
+
+For fixes and features, there might be additional branches, likely prefixed by `feature/` `fix/` followed by an issue number (if applicable) and/or a title.
+Feel free to adapt these naming scheme to your forks.
+
+### Merge Requirements
+
+To be merged into the master branch, your code has to pass the automated continuous integration tests, to ensure compatibility.
+In addition, your code has to be approved by a project member.
+
+#### What if my code fails the tests?
+
+Don't worry, you can submit your PR anyway. 
+The reviewing process might help you to solve remaining issues.
+
+### Commit messages
+
+Please use speaking titles and messages for your commits, to ensure a transparent history.
+If your patch fixes an issue, reference the ID in the first line.
+If you feel like you have to _briefly_ explain your changes, do it (for long explanations and discussion, consider opening an issue or describe in the PR).
+
+**Example commit:**
+```text
+Fix nasty bug from #1337
+
+This example commit fixes the issue that some people write non-speaking commit messages like 'done magic'.
+A short description is helpful sometimes.
+```
+
+You might sign your work, although that's no must.
+
+
+### When will it be merged?
+
+Short answer: When it makes sense.
+
+Bug fixes should be merged in time - assuming they pass the above criteria.
+New features might be assigned to a certain milestone and as a result of this be scheduled according to the planned release cycle.
+
+
+## Versioning
+
+This projects tries to adapt the [Semantic Versioning](https://semver.org).
+In short, bug fixes without do not affect any compatibility will raise the third number only, new features will be reflected in the second number and any change breaking compatibility with the public API require raising the first number.
+
+If you have to make a decision for which version to go please keep this in mind.
+However, for most non-member commits this is mostly informative, as the decision will be made by the project team later.
+
+
+## Build Environment
+
+All you need to start off - besides your favorite IDE of course - is Java and Maven.
+The project requires Java 8 or higher, both OpenJDK and Oracle JDK are supported.
+All build steps are executed by calling e.g. `mvn clean package` in the project's root directory,
+
+
+## Unit Tests
+
+The Java code is tested by a set of JUnit tests.
+All test files are located in the `src/test` directory. 
+Files ending with `Test.java` will be automatically included into the test suite.
+
+
+## Continuous Integration
+
+Automated tests are run using [Travis CI](https://travis-ci.org/stklcode/juraclient) for every commit including pull requests.
+
+There is also a code quality analysis pushing results to [SonarCloud](https://sonarcloud.io/dashboard?id=de.stklcode.pubtrans%3Ajuraclient).
+Keep in mind that the ruleset is not yet perfect, so not every minor issue has to be fixed immediately.
+
+
+## Still Open Questions?
+
+If anything is still left unanswered, and you're unsure if you got it right, don't hesitate to contact a team member.
+In any case you might submit your request/issue anyway, we won't refuse good code only for formal reasons.

README.md

@@ -1,8 +1,9 @@
 # jURAclient 
 [![Build status](https://travis-ci.org/stklcode/juraclient.svg?branch=master)](https://travis-ci.org/stklcode/juraclient)
-[![Quality Gate](https://sonarcloud.io/api/badges/gate?key=de.stklcode.pubtrans%3Ajuraclient)](https://sonarcloud.io/dashboard?id=de.stklcode.pubtrans%3Ajuraclient) 
-[![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](https://github.com/stklcode/juraclient/blob/master/LICENSE.txt) 
+[![Quality Gate](https://sonarcloud.io/api/project_badges/measure?project=de.stklcode.pubtrans%3Ajuraclient&metric=alert_status)](https://sonarcloud.io/dashboard?id=de.stklcode.pubtrans%3Ajuraclient) 
+[![Javadocs](https://www.javadoc.io/badge/de.stklcode.pubtrans/juraclient.svg)](https://www.javadoc.io/doc/de.stklcode.pubtrans/juraclient)
 [![Maven Central](https://img.shields.io/maven-central/v/de.stklcode.pubtrans/juraclient.svg)](https://search.maven.org/#search%7Cga%7C1%7Cg%3A%22de.stklcode.pubtrans%22%20AND%20a%3A%22juraclient%22)
+[![License](https://img.shields.io/badge/license-Apache%202.0-blue.svg)](https://github.com/stklcode/juraclient/blob/master/LICENSE.txt)
 
 Java client for URA based public transport APIs.
 
@@ -16,7 +17,7 @@ local bus station or any other custom queries. API versions 1.x and 2.x are supp
 // Instantiate the client (e.g. using the TFL API)
 UraClient ura = new UraClient("http://countdown.api.tfl.gov.uk");
 
-// Initiailize the API with non-standard endpoints (e.g. ASEAG with API V2)
+// Initialize the API with non-standard endpoints (e.g. ASEAG with API V2)
 UraClient ura = new UraClient("http://ivu.aseag.de", 
                               "interfaces/ura/instant_V2", 
                               "interfaces/ura/stream_V2");
@@ -38,31 +39,34 @@ List<Stop> stops = ura.forPosition(51.51009, -0.1345734, 200)
 
 ```java
 // Get next 10 trips for given stops and lines in a single direction (all filters optional)
-List<Trip> trips = ura.forStop("100000")
+List<Trip> trips = ura.forStops("100000")
                       .forLines("25", "35")
                       .forDirection(1)
                       .getTrips(10);
 
 // Get trips from given stop towards your destination
-List<Trip> trips = ura.forStopByName("Piccadilly Circus")
+List<Trip> trips = ura.forStopsByName("Piccadilly Circus")
                       .towards("Marble Arch")
                       .getTrips();
 ```
 
-## Maven Artifact
+### Get Messages
+
+```java
+// Get next 10 trips for given stops and lines in a single direction (all filters optional)
+List<Message> msgs = ura.forStops("100000")
+                        .getMessages();
 ```
+
+## Maven Artifact
+```xml
 <dependency>
     <groupId>de.stklcode.pubtrans</groupId>
     <artifactId>juraclient</artifactId>
-    <version>1.1.1</version>
+    <version>1.3.3</version>
 </dependency>
 ```
 
-## Planned Features
-
-* More refined query parameters
-* Stream API with asynchronous consumer
-
 ## License
 
 The project is licensed under [Apache License 2.0](http://www.apache.org/licenses/LICENSE-2.0).

pom.xml

@@ -1,12 +1,11 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
 
     <groupId>de.stklcode.pubtrans</groupId>
     <artifactId>juraclient</artifactId>
-    <version>1.1.1</version>
+    <version>1.3.3</version>
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
@@ -22,7 +21,7 @@
     <licenses>
         <license>
             <name>Apache License 2.0</name>
-            <url>http://www.apache.org/licenses/LICENSE-2.0.html</url>
+            <url>https://www.apache.org/licenses/LICENSE-2.0.html</url>
             <distribution>repo</distribution>
         </license>
     </licenses>
@@ -47,83 +46,91 @@
     </issueManagement>
 
     <dependencies>
-        <dependency>
-            <groupId>com.fasterxml.jackson.core</groupId>
-            <artifactId>jackson-core</artifactId>
-            <version>2.9.4</version>
-        </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.9.4</version>
+            <version>2.14.0</version>
         </dependency>
+
         <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>4.12</version>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>5.9.1</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.hamcrest</groupId>
-            <artifactId>hamcrest-junit</artifactId>
-            <version>2.0.0.0</version>
+            <artifactId>hamcrest</artifactId>
+            <version>2.2</version>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.powermock</groupId>
-            <artifactId>powermock-module-junit4</artifactId>
-            <version>2.0.0-beta.5</version>
-            <scope>test</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.powermock</groupId>
-            <artifactId>powermock-api-mockito2</artifactId>
-            <version>2.0.0-beta.5</version>
+            <groupId>com.github.tomakehurst</groupId>
+            <artifactId>wiremock-jre8</artifactId>
+            <version>2.35.0</version>
             <scope>test</scope>
         </dependency>
     </dependencies>
 
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.sonarsource.scanner.maven</groupId>
+                <artifactId>sonar-maven-plugin</artifactId>
+                <version>3.9.1.2184</version>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
     <build>
         <plugins>
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
-                <version>3.7.0</version>
+                <version>3.10.1</version>
                 <configuration>
                     <source>1.8</source>
                     <target>1.8</target>
                 </configuration>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-jar-plugin</artifactId>
+                <version>3.3.0</version>
+                <configuration>
+                    <archive>
+                        <manifest>
+                            <addClasspath>true</addClasspath>
+                            <addDefaultImplementationEntries>true</addDefaultImplementationEntries>
+                        </manifest>
+                        <manifestEntries>
+                            <Automatic-Module-Name>de.stklcode.pubtrans.juraclient</Automatic-Module-Name>
+                        </manifestEntries>
+                    </archive>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>2.22.2</version>
+            </plugin>
         </plugins>
     </build>
+
     <profiles>
         <profile>
-            <id>jacoco</id>
-            <dependencies>
-                <dependency>
-                    <groupId>org.jacoco</groupId>
-                    <artifactId>org.jacoco.agent</artifactId>
-                    <version>0.8.0</version>
-                    <classifier>runtime</classifier>
-                </dependency>
-            </dependencies>
+            <id>coverage</id>
             <build>
                 <plugins>
                     <plugin>
                         <groupId>org.jacoco</groupId>
                         <artifactId>jacoco-maven-plugin</artifactId>
-                        <version>0.8.0</version>
+                        <version>0.8.8</version>
                         <executions>
                             <execution>
-                                <id>default-instrument</id>
+                                <id>prepare-agent</id>
                                 <goals>
-                                    <goal>instrument</goal>
-                                </goals>
-                            </execution>
-                            <execution>
-                                <id>default-restore-instrumented-classes</id>
-                                <goals>
-                                    <goal>restore-instrumented-classes</goal>
+                                    <goal>prepare-agent</goal>
                                 </goals>
                             </execution>
                             <execution>
@@ -132,24 +139,116 @@
                                 <goals>
                                     <goal>report</goal>
                                 </goals>
-                                <configuration>
-                                    <dataFile>${project.build.directory}/coverage.exec</dataFile>
-                                </configuration>
                             </execution>
                         </executions>
                     </plugin>
-                    <plugin>
-                        <groupId>org.apache.maven.plugins</groupId>
-                        <artifactId>maven-surefire-plugin</artifactId>
-                        <version>2.20.1</version>
-                        <configuration>
-                            <systemPropertyVariables>
-                                <jacoco-agent.destfile>${project.build.directory}/jacoco.exec</jacoco-agent.destfile>
-                            </systemPropertyVariables>
-                        </configuration>
-                    </plugin>
                 </plugins>
             </build>
         </profile>
+
+        <profile>
+            <id>sources</id>
+            <activation>
+                <activeByDefault>false</activeByDefault>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-source-plugin</artifactId>
+                        <version>3.2.1</version>
+                        <executions>
+                            <execution>
+                                <id>attach-sources</id>
+                                <goals>
+                                    <goal>jar-no-fork</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+
+        <profile>
+            <id>javadoc</id>
+            <activation>
+                <activeByDefault>false</activeByDefault>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-javadoc-plugin</artifactId>
+                        <version>3.4.1</version>
+                        <configuration>
+                            <overview>${basedir}/src/main/javadoc/overview.html</overview>
+                            <source>1.8</source>
+                        </configuration>
+                        <executions>
+                            <execution>
+                                <id>attach-javadocs</id>
+                                <goals>
+                                    <goal>jar</goal>
+                                </goals>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+
+        <profile>
+            <id>sign</id>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-gpg-plugin</artifactId>
+                        <version>3.0.1</version>
+                        <executions>
+                            <execution>
+                                <id>sign-artifacts</id>
+                                <phase>verify</phase>
+                                <goals>
+                                    <goal>sign</goal>
+                                </goals>
+                                <configuration>
+                                    <keyname>${gpg.keyname}</keyname>
+                                </configuration>
+                            </execution>
+                        </executions>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+
+        <profile>
+            <id>local</id>
+            <distributionManagement>
+                <repository>
+                    <id>local</id>
+                    <url>${dist.repo.local}</url>
+                </repository>
+                <snapshotRepository>
+                    <id>local</id>
+                    <url>${dist.repo.local.snapshot}</url>
+                </snapshotRepository>
+            </distributionManagement>
+        </profile>
+
+        <profile>
+            <id>sonatype</id>
+            <distributionManagement>
+                <repository>
+                    <id>ossrh</id>
+                    <url>https://oss.sonatype.org/service/local/staging/deploy/maven2/</url>
+                </repository>
+                <snapshotRepository>
+                    <id>ossrh</id>
+                    <url>https://oss.sonatype.org/content/repositories/snapshots</url>
+                </snapshotRepository>
+            </distributionManagement>
+        </profile>
     </profiles>
 </project>

src/main/java/de/stklcode/pubtrans/ura/UraClient.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 Stefan Kalscheuer
+ * Copyright 2016-2022 Stefan Kalscheuer
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,23 +17,31 @@
 package de.stklcode.pubtrans.ura;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import de.stklcode.pubtrans.ura.model.Message;
 import de.stklcode.pubtrans.ura.model.Stop;
 import de.stklcode.pubtrans.ura.model.Trip;
+import de.stklcode.pubtrans.ura.reader.AsyncUraTripReader;
 
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
+import java.io.*;
 import java.net.URL;
+import java.net.URLEncoder;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.function.Consumer;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
 
 /**
  * Client for URA based public transport API.
+ * <p>
+ * This client features builder pattern style query functionality to obtain {@link Trip} and {@link Stop} information.
  *
  * @author Stefan Kalscheuer
  */
-public class UraClient {
+public class UraClient implements Serializable {
+    private static final long serialVersionUID = -1183740075816686611L;
+
     private static final String DEFAULT_INSTANT_URL = "/interfaces/ura/instant_V1";
     private static final String DEFAULT_STREAM_URL = "/interfaces/ura/stream_V1";
 
@@ -53,14 +61,21 @@ public class UraClient {
     private static final String PAR_ESTTIME = "EstimatedTime";
     private static final String PAR_TOWARDS = "Towards";
     private static final String PAR_CIRCLE = "Circle";
+    private static final String PAR_MSG_UUID = "MessageUUID";
+    private static final String PAR_MSG_TYPE = "MessageType";
+    private static final String PAR_MSG_PRIORITY = "MessagePriority";
+    private static final String PAR_MSG_TEXT = "MessageText";
 
     private static final Integer RES_TYPE_STOP = 0;
     private static final Integer RES_TYPE_PREDICTION = 1;
+    private static final Integer RES_TYPE_FLEX_MESSAGE = 2;
     private static final Integer RES_TYPE_URA_VERSION = 4;
 
     private static final String[] REQUEST_STOP = {PAR_STOP_NAME, PAR_STOP_ID, PAR_STOP_INDICATOR, PAR_STOP_STATE, PAR_GEOLOCATION};
     private static final String[] REQUEST_TRIP = {PAR_STOP_NAME, PAR_STOP_ID, PAR_STOP_INDICATOR, PAR_STOP_STATE, PAR_GEOLOCATION,
             PAR_VISIT_NUMBER, PAR_LINE_ID, PAR_LINE_NAME, PAR_DIR_ID, PAR_DEST_NAME, PAR_DEST_TEXT, PAR_VEHICLE_ID, PAR_TRIP_ID, PAR_ESTTIME};
+    private static final String[] REQUEST_MESSAGE = {PAR_STOP_NAME, PAR_STOP_ID, PAR_STOP_INDICATOR, PAR_STOP_STATE, PAR_GEOLOCATION,
+            PAR_MSG_UUID, PAR_MSG_TYPE, PAR_MSG_PRIORITY, PAR_MSG_TEXT};
 
     private final String baseURL;
     private final String instantURL;
@@ -177,7 +192,7 @@ public class UraClient {
 
     /**
      * Get list of trips.
-     * If forStops() and/or forLines() has been called, those will be used as filter.
+     * If {@link #forStops(String...)} and/or {@link #forLines(String...)} has been called, those will be used as filter.
      *
      * @return List of trips.
      */
@@ -187,7 +202,7 @@ public class UraClient {
 
     /**
      * Get list of trips with limit.
-     * If forStops() and/or forLines() has been called, those will be used as filter.
+     * If {@link #forStops(String...)} and/or {@link #forLines(String...)} has been called, those will be used as filter.
      *
      * @param limit Maximum number of results.
      * @return List of trips.
@@ -221,7 +236,7 @@ public class UraClient {
             String version = null;
             String line = br.readLine();
             while (line != null && (limit == null || trips.size() < limit)) {
-                List l = mapper.readValue(line, List.class);
+                List<?> l = mapper.readValue(line, List.class);
                 /* Check if result exists and has correct response type */
                 if (l != null && !l.isEmpty()) {
                     if (l.get(0).equals(RES_TYPE_URA_VERSION)) {
@@ -238,10 +253,46 @@ public class UraClient {
         return trips;
     }
 
+    /**
+     * Get trips for given stopIDs and lineIDs using stream API and pass each result to given consumer.
+     *
+     * @param query    The query.
+     * @param consumer Consumer(s) for single trips.
+     * @return Trip reader.
+     * @throws IOException Error reading response.
+     * @see #getTripsStream(Query, List)
+     * @since 1.2.0
+     */
+    public AsyncUraTripReader getTripsStream(final Query query, final Consumer<Trip> consumer) throws IOException {
+        return getTripsStream(query, Collections.singletonList(consumer));
+    }
+
+    /**
+     * Get trips for given stopIDs and lineIDs using stream API and pass each result to given consumers.
+     *
+     * @param query     The query.
+     * @param consumers Consumer(s) for single trips.
+     * @return Trip reader.
+     * @throws IOException Error retrieving stream response.
+     * @since 1.2.0
+     */
+    public AsyncUraTripReader getTripsStream(final Query query, final List<Consumer<Trip>> consumers) throws IOException {
+        // Create the reader.
+        AsyncUraTripReader reader = new AsyncUraTripReader(
+                new URL(requestURL(baseURL + streamURL, REQUEST_TRIP, query)),
+                consumers
+        );
+
+        // Open the reader, i.e. start reading from API.
+        reader.open();
+
+        return reader;
+    }
+
     /**
      * Get list of stops without filters.
      *
-     * @return Lhe list.
+     * @return The list of stops.
      */
     public List<Stop> getStops() {
         return getStops(new Query());
@@ -260,7 +311,7 @@ public class UraClient {
              BufferedReader br = new BufferedReader(new InputStreamReader(is))) {
             String line;
             while ((line = br.readLine()) != null) {
-                List l = mapper.readValue(line, List.class);
+                List<?> l = mapper.readValue(line, List.class);
                 /* Check if result exists and has correct response type */
                 if (l != null && !l.isEmpty() && l.get(0).equals(RES_TYPE_STOP)) {
                     stops.add(new Stop(l));
@@ -272,6 +323,73 @@ public class UraClient {
         return stops;
     }
 
+    /**
+     * Get list of messages.
+     *
+     * @return List of messages.
+     * @since 1.3
+     */
+    public List<Message> getMessages() {
+        return getMessages(new Query(), null);
+    }
+
+
+    /**
+     * Get list of messages.
+     * If forStops() has been called, those will be used as filter.
+     *
+     * @param query The query.
+     * @return List of trips.
+     * @since 1.3
+     */
+    public List<Message> getMessages(final Query query) {
+        return getMessages(query, null);
+    }
+
+
+    /**
+     * Get list of messages with limit.
+     *
+     * @param limit Maximum number of results.
+     * @return List of trips.
+     * @since 1.3.3
+     */
+    public List<Message> getMessages(final Integer limit) {
+        return getMessages(new Query(), limit);
+    }
+
+    /**
+     * Get list of messages for given stopIDs with result limit.
+     *
+     * @param query The query.
+     * @param limit Maximum number of results.
+     * @return List of trips.
+     * @since 1.3
+     */
+    public List<Message> getMessages(final Query query, final Integer limit) {
+        List<Message> messages = new ArrayList<>();
+        try (InputStream is = requestInstant(REQUEST_MESSAGE, query);
+             BufferedReader br = new BufferedReader(new InputStreamReader(is))) {
+            String version = null;
+            String line = br.readLine();
+            while (line != null && (limit == null || messages.size() < limit)) {
+                List<?> l = mapper.readValue(line, List.class);
+                /* Check if result exists and has correct response type */
+                if (l != null && !l.isEmpty()) {
+                    if (l.get(0).equals(RES_TYPE_URA_VERSION)) {
+                        version = l.get(1).toString();
+                    } else if (l.get(0).equals(RES_TYPE_FLEX_MESSAGE)) {
+                        messages.add(new Message(l, version));
+                    }
+                }
+                line = br.readLine();
+            }
+        } catch (IOException e) {
+            throw new IllegalStateException("Failed to read from API", e);
+        }
+        return messages;
+    }
+
     /**
      * Issue request to instant endpoint and return input stream.
      *
@@ -281,36 +399,59 @@ public class UraClient {
      * @throws IOException on errors
      */
     private InputStream requestInstant(final String[] returnList, final Query query) throws IOException {
-        String urlStr = baseURL + instantURL + "?ReturnList=" + String.join(",", returnList);
+        return request(requestURL(baseURL + instantURL, returnList, query));
+    }
+
+    /**
+     * Build request URL from given parameters.
+     *
+     * @param endpointURL Endpoint URL.
+     * @param returnList  Fields to fetch.
+     * @param query       The query.
+     * @return The URL
+     * @throws IOException on errors
+     * @since 1.2.0
+     */
+    private String requestURL(final String endpointURL, final String[] returnList, final Query query) throws IOException {
+        String urlStr = endpointURL + "?ReturnList=" + String.join(",", returnList);
 
         if (query.stopIDs != null && query.stopIDs.length > 0) {
-            urlStr += "&" + PAR_STOP_ID + "=" + String.join(",", query.stopIDs);
+            urlStr += "&" + PAR_STOP_ID + "=" + URLEncoder.encode(String.join(",", query.stopIDs), UTF_8.name());
         }
         if (query.stopNames != null && query.stopNames.length > 0) {
-            urlStr += "&" + PAR_STOP_NAME + "=" + String.join(",", query.stopNames);
+            urlStr += "&" + PAR_STOP_NAME + "=" + URLEncoder.encode(String.join(",", query.stopNames), UTF_8.name());
         }
         if (query.lineIDs != null && query.lineIDs.length > 0) {
-            urlStr += "&" + PAR_LINE_ID + "=" + String.join(",", query.lineIDs);
+            urlStr += "&" + PAR_LINE_ID + "=" + URLEncoder.encode(String.join(",", query.lineIDs), UTF_8.name());
         }
         if (query.lineNames != null && query.lineNames.length > 0) {
-            urlStr += "&" + PAR_LINE_NAME + "=" + String.join(",", query.lineNames);
+            urlStr += "&" + PAR_LINE_NAME + "=" + URLEncoder.encode(String.join(",", query.lineNames), UTF_8.name());
         }
         if (query.direction != null) {
             urlStr += "&" + PAR_DIR_ID + "=" + query.direction;
         }
         if (query.destinationNames != null) {
-            urlStr += "&" + PAR_DEST_NAME + "=" + String.join(",", query.destinationNames);
+            urlStr += "&" + PAR_DEST_NAME + "=" + URLEncoder.encode(String.join(",", query.destinationNames), UTF_8.name());
         }
         if (query.towards != null) {
-            urlStr += "&" + PAR_TOWARDS + "=" + String.join(",", query.towards);
+            urlStr += "&" + PAR_TOWARDS + "=" + URLEncoder.encode(String.join(",", query.towards), UTF_8.name());
         }
         if (query.circle != null) {
-            urlStr += "&" + PAR_CIRCLE + "=" + String.join(",", query.circle);
+            urlStr += "&" + PAR_CIRCLE + "=" + URLEncoder.encode(query.circle, UTF_8.name());
         }
 
-        URL url = new URL(urlStr);
+        return urlStr;
+    }
 
-        return url.openStream();
+    /**
+     * Open given URL as InputStream.
+     *
+     * @param url The URL.
+     * @return Input Stream of results.
+     * @throws IOException Error opening connection or reading data.
+     */
+    private InputStream request(String url) throws IOException {
+        return new URL(url).openStream();
     }
 
     /**
@@ -436,5 +577,62 @@ public class UraClient {
         public List<Trip> getTrips() {
             return UraClient.this.getTrips(this);
         }
+
+        /**
+         * Get trips for set filters.
+         *
+         * @param limit Maximum number of results.
+         * @return List of matching trips.
+         * @since 1.3.3
+         */
+        public List<Trip> getTrips(final Integer limit) {
+            return UraClient.this.getTrips(this, limit);
+        }
+
+        /**
+         * Get trips for set filters.
+         *
+         * @param consumer Consumer for single trips.
+         * @return Trip reader.
+         * @throws IOException Errors retrieving stream response.
+         * @see #getTripsStream(List)
+         * @since 1.2.0
+         */
+        public AsyncUraTripReader getTripsStream(Consumer<Trip> consumer) throws IOException {
+            return UraClient.this.getTripsStream(this, consumer);
+        }
+
+        /**
+         * Get trips for set filters.
+         *
+         * @param consumers Consumers for single trips.
+         * @return Trip reader.
+         * @throws IOException Errors retrieving stream response.
+         * @since 1.2.0
+         */
+        public AsyncUraTripReader getTripsStream(List<Consumer<Trip>> consumers) throws IOException {
+            return UraClient.this.getTripsStream(this, consumers);
+        }
+
+        /**
+         * Get trips for set filters.
+         *
+         * @return List of matching messages.
+         * @since 1.3
+         */
+        public List<Message> getMessages() {
+            return UraClient.this.getMessages(this);
+        }
+
+        /**
+         * Get trips for set filters with limit.
+         *
+         * @param limit Maximum number of results.
+         * @return List of matching messages.
+         * @since 1.3.3
+         */
+        public List<Message> getMessages(final Integer limit) {
+            return UraClient.this.getMessages(this, limit);
+        }
     }
 }

src/main/java/de/stklcode/pubtrans/ura/model/Message.java

@@ -0,0 +1,167 @@
+package de.stklcode.pubtrans.ura.model;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Entity for a message.
+ *
+ * @author Stefan Kalscheuer
+ * @since 1.3
+ */
+public class Message implements Model {
+    private static final long serialVersionUID = 5233610751062774273L;
+
+    private static final int MSG_UUID = 7;
+    private static final int MSG_TYPE = 8;
+    private static final int MSG_PRIORITY = 9;
+    private static final int MSG_TEXT = 10;
+    private static final int NUM_OF_FIELDS = 11;
+
+    private final Stop stop;
+    private final String uuid;
+    private final Integer type;
+    private final Integer priority;
+    private final String text;
+
+    /**
+     * Construct Message object from complete set of data.
+     *
+     * @param stopID        Stop ID.
+     * @param stopName      Stop name.
+     * @param stopIndicator Stop Indicator.
+     * @param stopState     Stop state.
+     * @param stopLatitude  Stop geolocation latitude.
+     * @param stopLongitude Stop geolocation latitude.
+     * @param msgUUID       Message UUID.
+     * @param msgType       Message type.
+     * @param msgPriority   Message priority.
+     * @param msgText       Message text.
+     */
+    public Message(final String stopID,
+                   final String stopName,
+                   final String stopIndicator,
+                   final Integer stopState,
+                   final Double stopLatitude,
+                   final Double stopLongitude,
+                   final String msgUUID,
+                   final Integer msgType,
+                   final Integer msgPriority,
+                   final String msgText) {
+        this(new Stop(stopID,
+                        stopName,
+                        stopIndicator,
+                        stopState,
+                        stopLatitude,
+                        stopLongitude),
+                msgUUID,
+                msgType,
+                msgPriority,
+                msgText);
+    }
+
+    /**
+     * Construct Message object from Stop model and set of additional data.
+     *
+     * @param stop        Stop model
+     * @param msgUUID     Message UUID.
+     * @param msgType     Message type.
+     * @param msgPriority Message priority.
+     * @param msgText     Message text.
+     */
+    public Message(final Stop stop,
+                   final String msgUUID,
+                   final Integer msgType,
+                   final Integer msgPriority,
+                   final String msgText) {
+        this.stop = stop;
+        this.uuid = msgUUID;
+        this.type = msgType;
+        this.priority = msgPriority;
+        this.text = msgText;
+    }
+
+    /**
+     * Construct Message object from raw list of attributes parsed from JSON.
+     *
+     * @param raw List of attributes from JSON line
+     * @throws IOException Thrown on invalid line format.
+     */
+    public Message(final List<?> raw) throws IOException {
+        this(raw, null);
+    }
+
+    /**
+     * Construct Message object from raw list of attributes parsed from JSON with explicitly specified version.
+     *
+     * @param raw     List of attributes from JSON line
+     * @param version API version
+     * @throws IOException Thrown on invalid line format.
+     */
+    public Message(final List<?> raw, final String version) throws IOException {
+        if (raw == null || raw.size() < NUM_OF_FIELDS) {
+            throw new IOException("Invalid number of fields");
+        }
+
+        stop = new Stop(raw);
+
+        if (raw.get(MSG_UUID) instanceof String) {
+            uuid = (String) raw.get(MSG_UUID);
+        } else {
+            throw Model.typeErrorString(MSG_UUID, raw.get(MSG_UUID).getClass());
+        }
+
+        if (raw.get(MSG_TYPE) instanceof Integer) {
+            type = (Integer) raw.get(MSG_TYPE);
+        } else {
+            throw Model.typeError(MSG_TYPE, raw.get(MSG_TYPE).getClass(), "Integer");
+        }
+
+        if (raw.get(MSG_PRIORITY) instanceof Integer) {
+            priority = (Integer) raw.get(MSG_PRIORITY);
+        } else {
+            throw Model.typeError(MSG_PRIORITY, raw.get(MSG_PRIORITY).getClass(), "Integer");
+        }
+
+        if (raw.get(MSG_TEXT) instanceof String) {
+            text = (String) raw.get(MSG_TEXT);
+        } else {
+            throw Model.typeErrorString(MSG_TEXT, raw.get(MSG_TEXT).getClass());
+        }
+    }
+
+    /**
+     * @return The affected stop.
+     */
+    public Stop getStop() {
+        return stop;
+    }
+
+    /**
+     * @return Message's unique identifier.
+     */
+    public String getUuid() {
+        return uuid;
+    }
+
+    /**
+     * @return Message type.
+     */
+    public Integer getType() {
+        return type;
+    }
+
+    /**
+     * @return Message priority. Lower value equals higher priority.
+     */
+    public Integer getPriority() {
+        return priority;
+    }
+
+    /**
+     * @return Message text.
+     */
+    public String getText() {
+        return text;
+    }
+}

src/main/java/de/stklcode/pubtrans/ura/model/Model.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 Stefan Kalscheuer
+ * Copyright 2016-2022 Stefan Kalscheuer
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,12 +17,15 @@
 package de.stklcode.pubtrans.ura.model;
 
 import java.io.IOException;
+import java.io.Serializable;
 
 /**
+ * Interface for model classes to bundle common methods.
+ *
  * @author Stefan Kalscheuer
  * @since 1.1.1
  */
-interface Model {
+interface Model extends Serializable {
     /**
      * Generate exception for unmatched type when String is expected.
      *
@@ -30,7 +33,7 @@ interface Model {
      * @param actual Actual class.
      * @return The Exception.
      */
-    static IOException typeErrorString(int field, Class actual) {
+    static IOException typeErrorString(int field, Class<?> actual) {
         return typeError(field, actual, "String");
     }
 
@@ -42,7 +45,7 @@ interface Model {
      * @param expected Expected type.
      * @return The Exception.
      */
-    static IOException typeError(int field, Class actual, String expected) {
+    static IOException typeError(int field, Class<?> actual, String expected) {
         return new IOException(String.format("Field %d not of expected type %s, found %s",
                 field, expected, actual.getSimpleName()));
     }

src/main/java/de/stklcode/pubtrans/ura/model/Stop.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 Stefan Kalscheuer
+ * Copyright 2016-2022 Stefan Kalscheuer
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,6 +25,8 @@ import java.util.List;
  * @author Stefan Kalscheuer
  */
 public final class Stop implements Model {
+    private static final long serialVersionUID = 202040044477267787L;
+
     private static final int F_STOP_NAME = 1;
     private static final int F_STOP_ID = 2;
     private static final int F_INDICATOR = 3;
@@ -70,7 +72,7 @@ public final class Stop implements Model {
      * @param raw List of attributes from JSON line
      * @throws IOException Thrown on invalid line format.
      */
-    public Stop(final List raw) throws IOException {
+    public Stop(final List<?> raw) throws IOException {
         if (raw == null || raw.size() < F_NUM_OF_FIELDS) {
             throw new IOException("Invalid number of fields");
         }
@@ -143,7 +145,7 @@ public final class Stop implements Model {
     }
 
     /**
-     * @return The stop geoloaction latitude.
+     * @return The stop geolocation latitude.
      */
     public Double getLatitude() {
         return latitude;

src/main/java/de/stklcode/pubtrans/ura/model/Trip.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 Stefan Kalscheuer
+ * Copyright 2016-2022 Stefan Kalscheuer
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -25,6 +25,8 @@ import java.util.List;
  * @author Stefan Kalscheuer
  */
 public final class Trip implements Model {
+    private static final long serialVersionUID = 7477381188869237381L;
+
     private static final int VISIT_ID = 7;
     private static final int LINE_ID = 8;
     private static final int LINE_NAME = 9;
@@ -140,18 +142,18 @@ public final class Trip implements Model {
      * @param raw List of attributes from JSON line
      * @throws IOException Thrown on invalid line format.
      */
-    public Trip(final List raw) throws IOException {
+    public Trip(final List<?> raw) throws IOException {
         this(raw, null);
     }
 
     /**
-     * Construct Stop object from raw list of attributes parsed from JSON with explicitly specified version.
+     * Construct Trip object from raw list of attributes parsed from JSON with explicitly specified version.
      *
      * @param raw     List of attributes from JSON line
      * @param version API version
      * @throws IOException Thrown on invalid line format.
      */
-    public Trip(final List raw, final String version) throws IOException {
+    public Trip(final List<?> raw, final String version) throws IOException {
         if (raw == null || raw.size() < NUM_OF_FIELDS) {
             throw new IOException("Invalid number of fields");
         }
@@ -176,13 +178,15 @@ public final class Trip implements Model {
             throw Model.typeErrorString(LINE_NAME, raw.get(LINE_NAME).getClass());
         }
 
-        if (raw.get(DIRECTION_ID) instanceof Integer) {
-            directionID = (Integer) raw.get(DIRECTION_ID);
+        if (raw.get(DIRECTION_ID) instanceof String                 // Also accept Strings (#2)
+                || raw.get(DIRECTION_ID) instanceof Integer
+                || raw.get(DIRECTION_ID) instanceof Long) {
+            directionID = Integer.valueOf(raw.get(DIRECTION_ID).toString());
             if (directionID < 0 || directionID > 2) {
                 throw new IOException("Direction out of range. Expected 1 or 2, found " + directionID);
             }
         } else {
-            throw Model.typeError(DIRECTION_ID, raw.get(DIRECTION_ID).getClass(), "Integer");
+            throw Model.typeError(DIRECTION_ID, raw.get(DIRECTION_ID).getClass(), "String/Long/Integer");
         }
 
         if (raw.get(DESTINATION_NAME) instanceof String) {
@@ -202,6 +206,8 @@ public final class Trip implements Model {
                 || raw.get(VEHICLE_ID) instanceof Integer
                 || raw.get(VEHICLE_ID) instanceof Long) {
             vehicleID = raw.get(VEHICLE_ID).toString();
+        } else if (raw.get(VEHICLE_ID) == null) {   // Only fail of field is not NULL (#3).
+            vehicleID = null;
         } else {
             throw Model.typeError(VEHICLE_ID, raw.get(VEHICLE_ID).getClass(), "String/Integer/Long");
         }
@@ -285,7 +291,7 @@ public final class Trip implements Model {
     }
 
     /**
-     * @return The vehicle ID.
+     * @return The vehicle ID or {@code null} if not present.
      */
     public String getVehicleID() {
         return vehicleID;

src/main/java/de/stklcode/pubtrans/ura/model/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2016-2022 Stefan Kalscheuer
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * jURAclient model classes that represent responses returned by the URA interface.
+ */
+package de.stklcode.pubtrans.ura.model;

src/main/java/de/stklcode/pubtrans/ura/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2016-2022 Stefan Kalscheuer
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * jURAclient base package - see {@link de.stklcode.pubtrans.ura.UraClient} for usage.
+ */
+package de.stklcode.pubtrans.ura;

src/main/java/de/stklcode/pubtrans/ura/reader/AsyncUraTripReader.java

@@ -0,0 +1,152 @@
+/*
+ * Copyright 2016-2022 Stefan Kalscheuer
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package de.stklcode.pubtrans.ura.reader;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import de.stklcode.pubtrans.ura.model.Trip;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.*;
+import java.util.function.Consumer;
+
+/**
+ * Asynchronous stream reader for URA stream API.
+ * <p>
+ * This reader provides a handler for asynchronous stream events.
+ *
+ * @author Stefan Kalscheuer
+ * @since 1.2.0
+ */
+public class AsyncUraTripReader implements AutoCloseable {
+    private static final Integer RES_TYPE_PREDICTION = 1;
+    private static final Integer RES_TYPE_URA_VERSION = 4;
+
+    private final List<Consumer<Trip>> consumers;
+    private final URL url;
+    private CompletableFuture<Void> future;
+    private boolean canceled;
+
+    /**
+     * Initialize trip reader.
+     *
+     * @param url      URL to read trips from.
+     * @param consumer Initial consumer.
+     */
+    public AsyncUraTripReader(URL url, Consumer<Trip> consumer) {
+        this.url = url;
+        this.consumers = new ArrayList<>();
+        this.consumers.add(consumer);
+    }
+
+    /**
+     * Initialize trip reader.
+     *
+     * @param url       URL to read trips from.
+     * @param consumers Initial list of consumers.
+     */
+    public AsyncUraTripReader(URL url, List<Consumer<Trip>> consumers) {
+        this.url = url;
+        this.consumers = new ArrayList<>(consumers);
+    }
+
+    public void open() {
+        // Throw exception, if future is already present.
+        if (future != null) {
+            throw new IllegalStateException("Reader already opened");
+        }
+
+        this.future = CompletableFuture.runAsync(() -> {
+            ObjectMapper mapper = new ObjectMapper();
+
+            try (InputStream is = getInputStream(url);
+                 BufferedReader br = new BufferedReader(new InputStreamReader(is))) {
+                String version = null;
+                String line = br.readLine();
+                while (line != null && !this.canceled) {
+                    List<?> l = mapper.readValue(line, List.class);
+                    // Check if result exists and has correct response type.
+                    if (l != null && !l.isEmpty()) {
+                        if (l.get(0).equals(RES_TYPE_URA_VERSION)) {
+                            version = l.get(1).toString();
+                        } else if (l.get(0).equals(RES_TYPE_PREDICTION)) {
+                            // Parse Trip and pass to each consumer.
+                            Trip trip = new Trip(l, version);
+                            this.consumers.forEach(c -> c.accept(trip));
+                        }
+                    }
+                    line = br.readLine();
+                }
+            } catch (IOException e) {
+                throw new IllegalStateException("Failed to read from API", e);
+            }
+        });
+    }
+
+    /**
+     * Register an additional consumer.
+     *
+     * @param consumer New consumer.
+     */
+    public void addConsumer(Consumer<Trip> consumer) {
+        consumers.add(consumer);
+    }
+
+    /**
+     * Close the reader.
+     * This is done by signaling cancel to the asynchronous task. If the task is not completed
+     * within 1 second however it is canceled hard.
+     */
+    @Override
+    public void close() {
+        // Nothing to do if future is not yet started.
+        if (future == null) {
+            return;
+        }
+
+        // Signal cancelling to gracefully stop future.
+        canceled = true;
+        try {
+            future.get(1, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        } catch (ExecutionException e) {
+            throw new IllegalStateException("Failed to read from API", e);
+        } catch (TimeoutException e) {
+            // Task failed to finish within 1 second.
+            future.cancel(true);
+        } finally {
+            future = null;
+        }
+    }
+
+    /**
+     * Get input stream from given URL.
+     *
+     * @param url URL to read from.
+     * @return Input Stream.
+     * @throws IOException On errors.
+     */
+    private static InputStream getInputStream(URL url) throws IOException {
+        return url.openStream();
+    }
+}

src/main/javadoc/overview.html

@@ -0,0 +1,14 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <title>API Overview</title>
+</head>
+<body>
+<p>Java client for URA based public transport APIs.</p>
+<p>
+    This client allows to simply connect any Java application to the public transport API to implement a monitor for the
+    local bus station or any other custom queries. API versions 1.x and 2.x are supported.
+</p>
+</body>
+</html>

src/test/java/de/stklcode/pubtrans/ura/UraClientTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 Stefan Kalscheuer
+ * Copyright 2016-2022 Stefan Kalscheuer
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,111 +16,115 @@
 
 package de.stklcode.pubtrans.ura;
 
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.client.WireMock;
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
+import de.stklcode.pubtrans.ura.model.Message;
 import de.stklcode.pubtrans.ura.model.Stop;
 import de.stklcode.pubtrans.ura.model.Trip;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.powermock.api.mockito.PowerMockito;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
-import java.io.InputStream;
-import java.net.URL;
 import java.util.List;
 import java.util.Optional;
 
+import static com.github.tomakehurst.wiremock.client.WireMock.*;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.CoreMatchers.startsWith;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.instanceOf;
-import static org.hamcrest.Matchers.nullValue;
 import static org.hamcrest.core.Is.is;
 
 /**
  * Unit test for the URA Client.
  * Tests run against mocked data collected from ASEAG API (http://ivu.aseag.de) and
- *  TFL API (http://http://countdown.api.tfl.gov.uk)
+ * TFL API (http://http://countdown.api.tfl.gov.uk)
  *
  * @author Stefan Kalscheuer
  */
-@RunWith(PowerMockRunner.class)
-@PrepareForTest({ UraClient.class, URL.class })
-public class UraClientTest {
-    @Test
-    public void getStopsTest() throws Exception {
-        /* Mock the HTTP call */
-        URL mockURL = PowerMockito.mock(URL.class);
-        PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL);
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V2_stops.txt"));
+class UraClientTest {
+    private static WireMockServer httpMock;
 
-        /* List stops and verify some values */
-        List<Stop> stops = new UraClient("mocked").getStops();
+    @BeforeAll
+    static void setUp() {
+        // Initialize HTTP mock.
+        httpMock = new WireMockServer(WireMockConfiguration.options().dynamicPort());
+        httpMock.start();
+        WireMock.configureFor("localhost", httpMock.port());
+    }
+
+    @AfterAll
+    static void tearDown() {
+        httpMock.stop();
+        httpMock = null;
+    }
+
+    @Test
+    void getStopsTest() {
+        // Mock the HTTP call.
+        mockHttpToFile(2, "instant_V2_stops.txt");
+
+        // List stops and verify some values.
+        List<Stop> stops = new UraClient(httpMock.baseUrl(), "/interfaces/ura/instant_V2", "/interfaces/ura/stream").getStops();
         assertThat(stops, hasSize(10));
         assertThat(stops.get(0).getId(), is("100210"));
         assertThat(stops.get(1).getName(), is("Brockenberg"));
-        assertThat(stops.get(2).getState(), is(0));;
+        assertThat(stops.get(2).getState(), is(0));
         assertThat(stops.get(3).getLatitude(), is(50.7578775));
         assertThat(stops.get(4).getLongitude(), is(6.0708663));
 
-        /* Test exception handling */
-        PowerMockito.when(mockURL.openStream()).thenReturn(new InputStream() {
-            @Override
-            public int read() throws IOException {
-                throw new IOException("Provoked exception 1.");
-            }
-        });
+        // Test Exception handling.
+        mockHttpToError(500);
+
         try {
-            new UraClient("mocked").getStops();
+            new UraClient(httpMock.baseUrl()).getStops();
         } catch (RuntimeException e) {
             assertThat(e, is(instanceOf(IllegalStateException.class)));
             assertThat(e.getCause(), is(instanceOf(IOException.class)));
-            assertThat(e.getCause().getMessage(), is("Provoked exception 1."));
+            assertThat(e.getCause().getMessage(), startsWith("Server returned HTTP response code: 500 for URL"));
         }
     }
 
     @Test
-    public void getStopsForLineTest() throws Exception {
-        /* Mock the HTTP call */
-        URL mockURL = PowerMockito.mock(URL.class);
-        PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL);
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V2_stops_line.txt"));
+    void getStopsForLineTest() {
+        // Mock the HTTP call.
+        mockHttpToFile(2, "instant_V2_stops_line.txt");
 
-        /* List stops and verify some values */
-        List<Stop> stops = new UraClient("mocked").forLines("33").getStops();
+        // List stops and verify some values.
+        List<Stop> stops = new UraClient(httpMock.baseUrl(), "/interfaces/ura/instant_V2", "/interfaces/ura/stream")
+                .forLines("33")
+                .getStops();
         assertThat(stops, hasSize(47));
         assertThat(stops.get(0).getId(), is("100000"));
         assertThat(stops.get(1).getName(), is("Kuckelkorn"));
-        assertThat(stops.get(2).getState(), is(0));;
+        assertThat(stops.get(2).getState(), is(0));
         assertThat(stops.get(3).getLatitude(), is(50.7690688));
         assertThat(stops.get(4).getIndicator(), is("H.1"));
         assertThat(stops.get(5).getLongitude(), is(6.2314072));
     }
 
     @Test
-    public void getStopsForPositionTest() throws Exception {
-        /* Mock the HTTP call */
-        URL mockURL = PowerMockito.mock(URL.class);
-        PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL);
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_stops_circle.txt"));
+    void getStopsForPositionTest() {
+        // Mock the HTTP call.
+        mockHttpToFile(1, "instant_V1_stops_circle.txt");
 
-        /* List stops and verify some values */
-        List<Stop> stops = new UraClient("mocked")
+        // List stops and verify some values.
+        List<Stop> stops = new UraClient(httpMock.baseUrl())
                 .forPosition(51.51009, -0.1345734, 200)
                 .getStops();
         assertThat(stops, hasSize(13));
         assertThat(stops.get(0).getId(), is("156"));
         assertThat(stops.get(1).getName(), is("Piccadilly Circus"));
-        assertThat(stops.get(2).getState(), is(0));;
+        assertThat(stops.get(2).getState(), is(0));
         assertThat(stops.get(3).getLatitude(), is(51.509154));
         assertThat(stops.get(4).getLongitude(), is(-0.134172));
         assertThat(stops.get(5).getIndicator(), is(nullValue()));
 
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_stops_circle_name.txt"));
-        stops = new UraClient("mocked")
+        mockHttpToFile(1, "instant_V1_stops_circle_name.txt");
+        stops = new UraClient(httpMock.baseUrl())
                 .forStopsByName("Piccadilly Circus")
                 .forPosition(51.51009, -0.1345734, 200)
                 .getStops();
@@ -129,22 +133,18 @@ public class UraClientTest {
     }
 
     @Test
-    public void getTripsForDestinationNamesTest() throws Exception {
-        /* Mock the HTTP call */
-        URL mockURL = PowerMockito.mock(URL.class);
-        PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL);
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_destination.txt"));
+    void getTripsForDestinationNamesTest() {
+        // Mock the HTTP call.
+        mockHttpToFile(1, "instant_V1_trips_destination.txt");
 
-        /* List stops and verify some values */
-        List<Trip> trips = new UraClient("mocked").forDestinationNames("Piccadilly Circus").getTrips();
+        // List stops and verify some values.
+        List<Trip> trips = new UraClient(httpMock.baseUrl()).forDestinationNames("Piccadilly Circus").getTrips();
         assertThat(trips, hasSize(9));
         assertThat(trips.stream().filter(t -> !t.getDestinationName().equals("Piccadilly Cir")).findAny(),
                 is(Optional.empty()));
 
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_stop_destination.txt"));
-        trips = new UraClient("mocked")
+        mockHttpToFile(1, "instant_V1_trips_stop_destination.txt");
+        trips = new UraClient(httpMock.baseUrl())
                 .forStops("156")
                 .forDestinationNames("Marble Arch")
                 .getTrips();
@@ -156,38 +156,31 @@ public class UraClientTest {
     }
 
     @Test
-    public void getTripsTowardsTest() throws Exception {
-        /* Mock the HTTP call */
-        URL mockURL = PowerMockito.mock(URL.class);
-        PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL);
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_towards.txt"));
+    void getTripsTowardsTest() {
+        // Mock the HTTP call.
+        mockHttpToFile(1, "instant_V1_trips_towards.txt");
 
         /* List stops and verify some values */
-        List<Trip> trips = new UraClient("mocked").towards("Marble Arch").getTrips();
+        List<Trip> trips = new UraClient(httpMock.baseUrl()).towards("Marble Arch").getTrips();
         assertThat(trips, hasSize(10));
 
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_stop_towards.txt"));
-        trips = new UraClient("mocked").forStops("156").towards("Marble Arch").getTrips();
+        mockHttpToFile(1, "instant_V1_trips_stop_towards.txt");
+        trips = new UraClient(httpMock.baseUrl()).forStops("156").towards("Marble Arch").getTrips();
         assertThat(trips, hasSize(17));
         assertThat(trips.stream().filter(t -> !t.getStop().getId().equals("156")).findAny(), is(Optional.empty()));
     }
 
     @Test
-    public void getTripsTest() throws Exception {
-        /* Mock the HTTP call */
-        URL mockURL = PowerMockito.mock(URL.class);
-        PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL);
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_all.txt"));
+    void getTripsTest() {
+        // Mock the HTTP call.
+        mockHttpToFile(1, "instant_V1_trips_all.txt");
 
-        /* Get trips without filters and verify some values */
-        List<Trip> trips = new UraClient("mocked").getTrips();
+        // Get trips without filters and verify some values.
+        List<Trip> trips = new UraClient(httpMock.baseUrl()).getTrips();
         assertThat(trips, hasSize(10));
         assertThat(trips.get(0).getId(), is("27000165015001"));
         assertThat(trips.get(1).getLineID(), is("55"));
-        assertThat(trips.get(2).getLineName(), is("28"));;
+        assertThat(trips.get(2).getLineName(), is("28"));
         assertThat(trips.get(3).getDirectionID(), is(1));
         assertThat(trips.get(4).getDestinationName(), is("Verlautenheide Endstr."));
         assertThat(trips.get(5).getDestinationText(), is("Aachen Bushof"));
@@ -196,15 +189,22 @@ public class UraClientTest {
         assertThat(trips.get(8).getVisitID(), is(30));
         assertThat(trips.get(9).getStop().getId(), is("100002"));
 
-        /* Repeat test for API V2 */
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V2_trips_all.txt"));
-        /* Get trips without filters and verify some values */
-        trips = new UraClient("mocked").getTrips();
+        // With limit.
+        trips = new UraClient(httpMock.baseUrl()).getTrips(5);
+        assertThat(trips, hasSize(5));
+        trips = new UraClient(httpMock.baseUrl()).getTrips(11);
+        assertThat(trips, hasSize(10));
+
+        // Repeat test for API V2.
+        mockHttpToFile(2, "instant_V2_trips_all.txt");
+
+        // Get trips without filters and verify some values.
+        trips = new UraClient(httpMock.baseUrl(), "/interfaces/ura/instant_V2", "/interfaces/ura/stream")
+                .getTrips();
         assertThat(trips, hasSize(10));
         assertThat(trips.get(0).getId(), is("27000165015001"));
         assertThat(trips.get(1).getLineID(), is("55"));
-        assertThat(trips.get(2).getLineName(), is("28"));;
+        assertThat(trips.get(2).getLineName(), is("28"));
         assertThat(trips.get(3).getDirectionID(), is(1));
         assertThat(trips.get(4).getDestinationName(), is("Verlautenheide Endstr."));
         assertThat(trips.get(5).getDestinationText(), is("Aachen Bushof"));
@@ -213,51 +213,47 @@ public class UraClientTest {
         assertThat(trips.get(8).getVisitID(), is(30));
         assertThat(trips.get(9).getStop().getId(), is("100002"));
 
-        /* Get limited number of trips */
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_all.txt"));
-        trips = new UraClient("mocked").getTrips(5);
+        // Get limited number of trips.
+        mockHttpToFile(1, "instant_V1_trips_all.txt");
+        trips = new UraClient(httpMock.baseUrl()).getTrips(5);
         assertThat(trips, hasSize(5));
 
-        /* Test exception handling */
-        PowerMockito.when(mockURL.openStream()).thenReturn(new InputStream() {
-            @Override
-            public int read() throws IOException {
-                throw new IOException("Provoked exception 1.");
-            }
-        });
+        // Test mockException handling.
+        mockHttpToError(502);
         try {
-            new UraClient("mocked").getTrips();
+            new UraClient(httpMock.baseUrl()).getTrips();
         } catch (RuntimeException e) {
             assertThat(e, is(instanceOf(IllegalStateException.class)));
             assertThat(e.getCause(), is(instanceOf(IOException.class)));
-            assertThat(e.getCause().getMessage(), is("Provoked exception 1."));
+            assertThat(e.getCause().getMessage(), startsWith("Server returned HTTP response code: 502 for URL"));
         }
     }
 
     @Test
-    public void getTripsForStopTest() throws Exception {
-        /* Mock the HTTP call */
-        URL mockURL = PowerMockito.mock(URL.class);
-        PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL);
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_stop.txt"));
+    void getTripsForStopTest() {
+        // Mock the HTTP call.
+        mockHttpToFile(1, "instant_V1_trips_stop.txt");
 
-        /* Get trips for stop ID 100000 (Aachen Bushof) and verify some values */
-        List<Trip> trips = new UraClient("mocked")
+        // Get trips for stop ID 100000 (Aachen Bushof) and verify some values.
+        List<Trip> trips = new UraClient(httpMock.baseUrl())
                 .forStops("100000")
                 .getTrips();
         assertThat(trips, hasSize(10));
         assertThat(trips.stream().filter(t -> !t.getStop().getId().equals("100000")).findAny(), is(Optional.empty()));
         assertThat(trips.get(0).getId(), is("27000158010001"));
         assertThat(trips.get(1).getLineID(), is("7"));
-        assertThat(trips.get(2).getLineName(), is("25"));;
+        assertThat(trips.get(2).getLineName(), is("25"));
         assertThat(trips.get(3).getStop().getIndicator(), is("H.15"));
 
-        /* Get trips for stop name "Uniklinik" and verify some values */
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_stop_name.txt"));
-        trips = new UraClient("mocked")
+        // With limit.
+        trips = new UraClient(httpMock.baseUrl())
+                .forStops("100000")
+                .getTrips(7);
+        assertThat(trips, hasSize(7));
+
+        // Get trips for stop name "Uniklinik" and verify some values.
+        mockHttpToFile(1, "instant_V1_trips_stop_name.txt");
+        trips = new UraClient(httpMock.baseUrl())
                 .forStopsByName("Uniklinik")
                 .getTrips();
         assertThat(trips, hasSize(10));
@@ -265,46 +261,41 @@ public class UraClientTest {
                 is(Optional.empty()));
         assertThat(trips.get(0).getId(), is("92000043013001"));
         assertThat(trips.get(1).getLineID(), is("5"));
-        assertThat(trips.get(2).getVehicleID(), is("317"));;
+        assertThat(trips.get(2).getVehicleID(), is("317"));
         assertThat(trips.get(3).getDirectionID(), is(1));
     }
 
     @Test
-    public void getTripsForLine() throws Exception {
-        /* Mock the HTTP call */
-        URL mockURL = PowerMockito.mock(URL.class);
-        PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL);
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_line.txt"));
+    void getTripsForLine() {
+        // Mock the HTTP call.
+        mockHttpToFile(1, "instant_V1_trips_line.txt");
 
-        /* Get trips for line ID 3 and verify some values */
-        List<Trip> trips = new UraClient("mocked")
+        // Get trips for line ID 3 and verify some values.
+        List<Trip> trips = new UraClient(httpMock.baseUrl())
                 .forLines("3")
                 .getTrips();
         assertThat(trips, hasSize(10));
         assertThat(trips.stream().filter(t -> !t.getLineID().equals("3")).findAny(), is(Optional.empty()));
         assertThat(trips.get(0).getId(), is("27000154004001"));
         assertThat(trips.get(1).getLineID(), is("3"));
-        assertThat(trips.get(2).getLineName(), is("3.A"));;
+        assertThat(trips.get(2).getLineName(), is("3.A"));
         assertThat(trips.get(3).getStop().getIndicator(), is("H.4 (Pontwall)"));
 
-        /* Get trips for line name "3.A" and verify some values */
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_line_name.txt"));
-        trips = new UraClient("mocked")
+        // Get trips for line name "3.A" and verify some values.
+        mockHttpToFile(1, "instant_V1_trips_line_name.txt");
+        trips = new UraClient(httpMock.baseUrl())
                 .forLinesByName("3.A")
                 .getTrips();
         assertThat(trips, hasSize(10));
         assertThat(trips.stream().filter(t -> !t.getLineName().equals("3.A")).findAny(), is(Optional.empty()));
         assertThat(trips.get(0).getId(), is("92000288014001"));
         assertThat(trips.get(1).getLineID(), is("3"));
-        assertThat(trips.get(2).getLineName(), is("3.A"));;
+        assertThat(trips.get(2).getLineName(), is("3.A"));
         assertThat(trips.get(3).getStop().getName(), is("Aachen Gartenstraße"));
 
-        /* Get trips for line 3 with direction 1 and verify some values */
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_line_direction.txt"));
-        trips = new UraClient("mocked")
+        // Get trips for line 3 with direction 1 and verify some values.
+        mockHttpToFile(1, "instant_V1_trips_line_direction.txt");
+        trips = new UraClient(httpMock.baseUrl())
                 .forLines("412")
                 .forDirection(2)
                 .getTrips();
@@ -312,10 +303,9 @@ public class UraClientTest {
         assertThat(trips.stream().filter(t -> !t.getLineID().equals("412")).findAny(), is(Optional.empty()));
         assertThat(trips.stream().filter(t -> !t.getDirectionID().equals(2)).findAny(), is(Optional.empty()));
 
-        /* Test lineID and direction in different order */
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_line_direction.txt"));
-        trips = new UraClient("mocked")
+        // Test lineID and direction in different order.
+        mockHttpToFile(1, "instant_V1_trips_line_direction.txt");
+        trips = new UraClient(httpMock.baseUrl())
                 .forDirection(2)
                 .forLines("412")
                 .getTrips();
@@ -325,15 +315,12 @@ public class UraClientTest {
     }
 
     @Test
-    public void getTripsForStopAndLine() throws Exception {
-        /* Mock the HTTP call */
-        URL mockURL = PowerMockito.mock(URL.class);
-        PowerMockito.whenNew(URL.class).withAnyArguments().thenReturn(mockURL);
-        PowerMockito.when(mockURL.openStream())
-                .thenReturn(getClass().getResourceAsStream("instant_V1_trips_stop_line.txt"));
+    void getTripsForStopAndLine() {
+        // Mock the HTTP call.
+        mockHttpToFile(1, "instant_V1_trips_stop_line.txt");
 
-        /* Get trips for line ID 25 and 25 at stop 100000 and verify some values */
-        List<Trip> trips = new UraClient("mocked")
+        // Get trips for line ID 25 and 25 at stop 100000 and verify some values.
+        List<Trip> trips = new UraClient(httpMock.baseUrl())
                 .forLines("25", "35")
                 .forStops("100000")
                 .getTrips();
@@ -343,7 +330,74 @@ public class UraClientTest {
         assertThat(trips.stream().filter(t -> !t.getStop().getId().equals("100000")).findAny(), is(Optional.empty()));
         assertThat(trips.get(0).getId(), is("27000078014001"));
         assertThat(trips.get(1).getLineID(), is("25"));
-        assertThat(trips.get(3).getLineName(), is("35"));;
+        assertThat(trips.get(3).getLineName(), is("35"));
         assertThat(trips.get(5).getStop().getIndicator(), is("H.12"));
     }
+
+
+    @Test
+    void getMessages() {
+        UraClient uraClient = new UraClient(httpMock.baseUrl());
+
+        // Mock the HTTP call.
+        mockHttpToFile(1, "instant_V1_messages.txt");
+
+        // Get messages without filter and verify some values.
+        List<Message> messages = uraClient.getMessages();
+        assertThat(messages, hasSize(2));
+        assertThat(messages.get(0).getStop().getId(), is("100707"));
+        assertThat(messages.get(0).getUuid(), is("016e1231d4e30014_100707"));
+        assertThat(messages.get(1).getStop().getName(), is("Herzogenr. Rathaus"));
+        assertThat(messages.get(1).getUuid(), is("016e2cc3a3750006_210511"));
+        assertThat(messages.get(0).getType(), is(0));
+        assertThat(messages.get(1).getPriority(), is(0));
+        assertThat(messages.get(0).getText(), is("Sehr geehrte Fahrgäste, wegen Strassenbauarbeiten kann diese Haltestelle nicht von den Bussen der Linien 17, 44 und N2 angefahren werden."));
+        assertThat(messages.get(1).getText(), is("Sehr geehrte Fahrgäste, diese Haltestelle wird vorübergehend von den Linien 47, 147 und N3 nicht angefahren."));
+
+        // With limit.
+        messages = uraClient.getMessages(1);
+        assertThat(messages, hasSize(1));
+        messages = uraClient.getMessages(3);
+        assertThat(messages, hasSize(2));
+    }
+
+    @Test
+    void getMessagesForStop() {
+        UraClient uraClient = new UraClient(httpMock.baseUrl(), "/interfaces/ura/instant_V2", "/interfaces/ura/stream");
+
+        // Mock the HTTP call.
+        mockHttpToFile(2, "instant_V2_messages_stop.txt");
+
+        // Get trips for stop ID 100707 (Berensberger Str.) and verify some values.
+        List<Message> messages = uraClient.forStops("100707").getMessages();
+        assertThat(messages, hasSize(1));
+        assertThat(messages.stream().filter(t -> !t.getStop().getId().equals("100707")).findAny(), is(Optional.empty()));
+        assertThat(messages.get(0).getUuid(), is("016e1231d4e30014_100707"));
+        assertThat(messages.get(0).getType(), is(0));
+        assertThat(messages.get(0).getPriority(), is(3));
+        assertThat(messages.get(0).getText(), is("Sehr geehrte Fahrgäste, wegen Strassenbauarbeiten kann diese Haltestelle nicht von den Bussen der Linien 17, 44 und N2 angefahren werden."));
+
+        // With limit.
+        messages = uraClient.forStops("100707").getMessages(0);
+        assertThat(messages, hasSize(0));
+        messages = uraClient.forStops("100707").getMessages(2);
+        assertThat(messages, hasSize(1));
+    }
+
+
+    private static void mockHttpToFile(int version, String resourceFile) {
+        WireMock.stubFor(
+                get(urlPathEqualTo("/interfaces/ura/instant_V" + version)).willReturn(
+                        aResponse().withBodyFile(resourceFile)
+                )
+        );
+    }
+
+    private static void mockHttpToError(int code) {
+        WireMock.stubFor(
+                get(anyUrl()).willReturn(
+                        aResponse().withStatus(code)
+                )
+        );
+    }
 }

src/test/java/de/stklcode/pubtrans/ura/model/MessageTest.java

@@ -0,0 +1,153 @@
+/*
+ * Copyright 2016-2022 Stefan Kalscheuer
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package de.stklcode.pubtrans.ura.model;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Unit test for the {@link Message} meta model.
+ *
+ * @author Stefan Kalscheuer
+ */
+public class MessageTest {
+    @Test
+    public void basicConstructorTest() {
+        Message message = new Message("sid",
+                "name",
+                "indicator",
+                1,
+                2.345,
+                6.789,
+                "msg_uuid",
+                1,
+                3,
+                "message text");
+        assertThat(message.getStop().getId(), is("sid"));
+        assertThat(message.getStop().getName(), is("name"));
+        assertThat(message.getStop().getIndicator(), is("indicator"));
+        assertThat(message.getStop().getState(), is(1));
+        assertThat(message.getStop().getLatitude(), is(2.345));
+        assertThat(message.getStop().getLongitude(), is(6.789));
+        assertThat(message.getUuid(), is("msg_uuid"));
+        assertThat(message.getType(), is(1));
+        assertThat(message.getPriority(), is(3));
+        assertThat(message.getText(), is("message text"));
+    }
+
+    @Test
+    public void listConstructorTest() {
+        /* Create valid raw data list */
+        List<Object> raw = new ArrayList<>();
+        raw.add(1);
+        raw.add("stopName");
+        raw.add("stopId");
+        raw.add("stopIndicator");
+        raw.add(9);
+        raw.add(8.765);
+        raw.add(43.21);
+        raw.add("msg_uuid");
+        raw.add(1);
+        raw.add(3);
+        raw.add("message text");
+
+        try {
+            Message message = new Message(raw);
+            assertThat(message.getStop().getId(), is("stopId"));
+            assertThat(message.getStop().getName(), is("stopName"));
+            assertThat(message.getStop().getIndicator(), is("stopIndicator"));
+            assertThat(message.getStop().getState(), is(9));
+            assertThat(message.getStop().getLatitude(), is(8.765));
+            assertThat(message.getStop().getLongitude(), is(43.21));
+            assertThat(message.getUuid(), is("msg_uuid"));
+            assertThat(message.getType(), is(1));
+            assertThat(message.getPriority(), is(3));
+            assertThat(message.getText(), is("message text"));
+        } catch (IOException e) {
+            fail("Creation of Message from valid list failed: " + e.getMessage());
+        }
+
+        /* Excess elements should be ignored */
+        raw.add("foo");
+        try {
+            Message message = new Message(raw);
+            assertThat(message, is(notNullValue()));
+            raw.remove(11);
+        } catch (IOException e) {
+            fail("Creation of Message from valid list failed: " + e.getMessage());
+        }
+
+        /* Test exceptions on invalid data */
+        List<Object> invalid = new ArrayList<>(raw);
+        invalid.remove(7);
+        invalid.add(7, 123L);
+        try {
+            new Message(invalid);
+            fail("Creation of Message with invalid UUID field successful");
+        } catch (Exception e) {
+            assertThat(e, is(instanceOf(IOException.class)));
+        }
+
+        invalid = new ArrayList<>(raw);
+        invalid.remove(8);
+        invalid.add(8, "abc");
+        try {
+            new Message(invalid);
+            fail("Creation of Message with invalid type field successful");
+        } catch (Exception e) {
+            assertThat(e, is(instanceOf(IOException.class)));
+        }
+
+        invalid = new ArrayList<>(raw);
+        invalid.remove(9);
+        invalid.add(9, "xyz");
+        try {
+            new Message(invalid);
+            fail("Creation of Message with invalid priority field successful");
+        } catch (Exception e) {
+            assertThat(e, is(instanceOf(IOException.class)));
+        }
+
+        invalid = new ArrayList<>(raw);
+        invalid.remove(10);
+        invalid.add(10, 1.23);
+        try {
+            new Message(invalid);
+            fail("Creation of Message with invalid text field successful");
+        } catch (Exception e) {
+            assertThat(e, is(instanceOf(IOException.class)));
+        }
+
+        invalid = new ArrayList<>(raw);
+        invalid.remove(10);
+        try {
+            new Message(invalid);
+            fail("Creation of Message with too short list successful");
+        } catch (Exception e) {
+            assertThat(e, is(instanceOf(IOException.class)));
+        }
+    }
+}

src/test/java/de/stklcode/pubtrans/ura/model/StopTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 Stefan Kalscheuer
+ * Copyright 2016-2022 Stefan Kalscheuer
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
 
 package de.stklcode.pubtrans.ura.model;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -26,10 +26,10 @@ import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Is.is;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.fail;
 
 /**
- * Unit test for the Stop metamodel.
+ * Unit test for the {@link Stop} meta model.
  *
  * @author Stefan Kalscheuer
  */
@@ -85,7 +85,7 @@ public class StopTest {
         invalid.add(1, 5);
         try {
             new Stop(invalid);
-            fail("Creation of Stop with invalid name field successfull");
+            fail("Creation of Stop with invalid name field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -95,7 +95,7 @@ public class StopTest {
         invalid.add(2, 0);
         try {
             new Stop(invalid);
-            fail("Creation of Stop with invalid id field successfull");
+            fail("Creation of Stop with invalid id field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -105,7 +105,7 @@ public class StopTest {
         invalid.add(3, -1.23);
         try {
             new Stop(invalid);
-            fail("Creation of Stop with invalid indicator field successfull");
+            fail("Creation of Stop with invalid indicator field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -115,7 +115,7 @@ public class StopTest {
         invalid.add(4, "foo");
         try {
             new Stop(invalid);
-            fail("Creation of Stop with invalid state field successfull");
+            fail("Creation of Stop with invalid state field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -125,7 +125,7 @@ public class StopTest {
         invalid.add(5, "123");
         try {
             new Stop(invalid);
-            fail("Creation of Stop with invalid latitude field successfull");
+            fail("Creation of Stop with invalid latitude field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -135,7 +135,7 @@ public class StopTest {
         invalid.add(6, 456);
         try {
             new Stop(invalid);
-            fail("Creation of Stop with invalid longitude field successfull");
+            fail("Creation of Stop with invalid longitude field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -144,7 +144,7 @@ public class StopTest {
         invalid.remove(6);
         try {
             new Stop(invalid);
-            fail("Creation of Stop with too short list successfull");
+            fail("Creation of Stop with too short list successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }

src/test/java/de/stklcode/pubtrans/ura/model/TripTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 Stefan Kalscheuer
+ * Copyright 2016-2022 Stefan Kalscheuer
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
 
 package de.stklcode.pubtrans.ura.model;
 
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -26,10 +26,10 @@ import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.core.Is.is;
-import static org.junit.Assert.fail;
+import static org.junit.jupiter.api.Assertions.fail;
 
 /**
- * Unit test for the Trip metamodel.
+ * Unit test for the {@link Trip} meta model.
  *
  * @author Stefan Kalscheuer
  */
@@ -130,13 +130,33 @@ public class TripTest {
             fail("Creation of Trip from valid list failed: " + e.getMessage());
         }
 
+        raw.remove(10);
+        raw.add(10, 0L);    // Long values are OK.
+        try {
+            Trip trip = new Trip(raw);
+            assertThat(trip, is(notNullValue()));
+            assertThat(trip.getDirectionID(), is(0));
+        } catch (IOException e) {
+            fail("Creation of Trip from valid list failed: " + e.getMessage());
+        }
+
+        raw.remove(10);
+        raw.add(10, "0");    // String values are OK.
+        try {
+            Trip trip = new Trip(raw);
+            assertThat(trip, is(notNullValue()));
+            assertThat(trip.getDirectionID(), is(0));
+        } catch (IOException e) {
+            fail("Creation of Trip from valid list failed: " + e.getMessage());
+        }
+
         /* Test exceptions on invalid data */
         List<Object> invalid = new ArrayList<>(raw);
         invalid.remove(7);
         invalid.add(7, "123");
         try {
             new Trip(invalid);
-            fail("Creation of Trip with invalid visitID field successfull");
+            fail("Creation of Trip with invalid visitID field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -146,7 +166,7 @@ public class TripTest {
         invalid.add(8, 25);
         try {
             new Trip(invalid);
-            fail("Creation of Trip with invalid lineID field successfull");
+            fail("Creation of Trip with invalid lineID field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -156,17 +176,17 @@ public class TripTest {
         invalid.add(9, 234L);
         try {
             new Trip(invalid);
-            fail("Creation of Trip with invalid line name field successfull");
+            fail("Creation of Trip with invalid line name field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
 
         invalid = new ArrayList<>(raw);
         invalid.remove(10);
-        invalid.add(10, "1");
+        invalid.add(10, "7");   // Strings are generally OK, but 7 is out of range (#2).
         try {
             new Trip(invalid);
-            fail("Creation of Trip with invalid directionID field successfull");
+            fail("Creation of Trip with invalid directionID field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -176,7 +196,7 @@ public class TripTest {
         invalid.add(11, 987);
         try {
             new Trip(invalid);
-            fail("Creation of Trip with invalid destinationName field successfull");
+            fail("Creation of Trip with invalid destinationName field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -186,7 +206,7 @@ public class TripTest {
         invalid.add(12, 456.78);
         try {
             new Trip(invalid);
-            fail("Creation of Trip with invalid destinationText field successfull");
+            fail("Creation of Trip with invalid destinationText field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -196,7 +216,7 @@ public class TripTest {
         invalid.add(13, 'x');
         try {
             new Trip(invalid);
-            fail("Creation of Trip with invalid vehicleID field successfull");
+            fail("Creation of Trip with invalid vehicleID field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -206,7 +226,7 @@ public class TripTest {
         invalid.add(14, 1.2);
         try {
             new Trip(invalid);
-            fail("Creation of Trip with invalid id field successfull");
+            fail("Creation of Trip with invalid id field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -216,7 +236,7 @@ public class TripTest {
         invalid.add(15, 456);
         try {
             new Trip(invalid);
-            fail("Creation of Trip with invalid estimatedTime field successfull");
+            fail("Creation of Trip with invalid estimatedTime field successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -225,7 +245,7 @@ public class TripTest {
         invalid.remove(15);
         try {
             new Trip(invalid);
-            fail("Creation of Trip with too short list successfull");
+            fail("Creation of Trip with too short list successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }
@@ -234,7 +254,7 @@ public class TripTest {
         invalid.set(10, 3);
         try {
             new Trip(invalid);
-            fail("Creation of Trip with direction ID 3 successfull");
+            fail("Creation of Trip with direction ID 3 successful");
         } catch (Exception e) {
             assertThat(e, is(instanceOf(IOException.class)));
         }

src/test/java/de/stklcode/pubtrans/ura/reader/AsyncUraTripReaderTest.java

@@ -0,0 +1,239 @@
+/*
+ * Copyright 2016-2022 Stefan Kalscheuer
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package de.stklcode.pubtrans.ura.reader;
+
+import com.github.tomakehurst.wiremock.WireMockServer;
+import com.github.tomakehurst.wiremock.client.WireMock;
+import com.github.tomakehurst.wiremock.common.FileSource;
+import com.github.tomakehurst.wiremock.core.WireMockConfiguration;
+import com.github.tomakehurst.wiremock.extension.Parameters;
+import com.github.tomakehurst.wiremock.extension.ResponseTransformer;
+import com.github.tomakehurst.wiremock.http.ChunkedDribbleDelay;
+import com.github.tomakehurst.wiremock.http.Request;
+import com.github.tomakehurst.wiremock.http.Response;
+import de.stklcode.pubtrans.ura.model.Trip;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.URL;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.*;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.Is.is;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+/**
+ * Unit test for the asynchronous URA Trip reader.
+ * <p>
+ * Because this test runs asynchronously, it might not work as expected in debugging environments.
+ * Stream input files are normalized to equal line length and split into chunks, one line each 500ms.
+ *
+ * @author Stefan Kalscheuer
+ */
+public class AsyncUraTripReaderTest {
+    private static WireMockServer httpMock;
+
+    @BeforeAll
+    public static void setUp() {
+        // Initialize HTTP mock.
+        httpMock = new WireMockServer(WireMockConfiguration.options().dynamicPort()
+                .asynchronousResponseEnabled(true)
+                .extensions(StreamTransformer.class)
+        );
+        httpMock.start();
+        WireMock.configureFor("localhost", httpMock.port());
+    }
+
+    @AfterAll
+    public static void tearDown() {
+        httpMock.stop();
+        httpMock = null;
+    }
+
+    /**
+     * Test the reader.
+     * This test contains some timing values, which is not very nice for debugging, but should do the job here
+     * as 1s is most likely more than enough time on any reasonable build system to parse some simple JSON lines.
+     *
+     * @throws InterruptedException Thread interrupted.
+     * @throws IOException          Error reading or writing mocked data.
+     */
+    @Test
+    public void readerTest() throws InterruptedException, IOException {
+        // Callback counter for some unhandy async mockery.
+        final AtomicInteger counter = new AtomicInteger(0);
+
+        // The list which will be populated by the callback.
+        Deque<Trip> trips = new ConcurrentLinkedDeque<>();
+
+        // Start with V1 data and read file to mock list.
+        readLinesToMock(1, "/__files/stream_V1_stops_all.txt", 8);
+
+        AsyncUraTripReader tr = new AsyncUraTripReader(
+                new URL(httpMock.baseUrl() + "/interfaces/ura/stream_V1"),
+                Collections.singletonList(
+                        trip -> {
+                            trips.add(trip);
+                            counter.incrementAndGet();
+                        }
+                )
+        );
+
+        // Open the reader.
+        tr.open();
+        // Read for 1 second.
+        TimeUnit.SECONDS.sleep(1);
+        assumeTrue(trips.isEmpty(), "Trips should empty after 1s without reading");
+
+        // Wait another 1s for the callback to be triggered.
+        TimeUnit.SECONDS.sleep(1);
+
+        assertThat("Unexpected number of trips after first entry", trips.size(), is(2));
+
+        // Flush all remaining lines.
+        TimeUnit.SECONDS.sleep(3);
+
+        assertThat("Unexpected number of trips after all lines have been flushed", trips.size(), is(7));
+
+        // Clear trip list and repeat with V2 data.
+        trips.clear();
+        readLinesToMock(2, "/__files/stream_V2_stops_all.txt", 8);
+
+        tr = new AsyncUraTripReader(
+                new URL(httpMock.baseUrl() + "/interfaces/ura/stream_V2"),
+                trips::add
+        );
+
+        // Open the reader.
+        tr.open();
+        // Read for 1 second.
+        TimeUnit.SECONDS.sleep(1);
+        assumeTrue(trips.isEmpty(), "Trips should empty after 1s without reading");
+
+        TimeUnit.SECONDS.sleep(1);
+        assertThat("Unexpected number of v2 trips after first entry", trips.size(), is(2));
+
+        // Add a second consumer that pushes to another list.
+        Deque<Trip> trips2 = new ConcurrentLinkedDeque<>();
+        tr.addConsumer(trips2::add);
+
+        // Flush all remaining lines.
+        TimeUnit.SECONDS.sleep(3);
+
+        tr.close();
+
+        assertThat("Unexpected number of v2 trips after all lines have been flushed", trips.size(), is(7));
+        assertThat("Unexpected number of v2 trips in list 2 after all lines have been flushed", trips2.size(), is(5));
+        assertThat("Same object should have been pushed to both lists", trips.containsAll(trips2));
+
+        // Opening the reader twice should raise an exception.
+        assertDoesNotThrow(tr::open, "Opening the reader after closing should not fail");
+        assertThrows(IllegalStateException.class, tr::open, "Opening the reader twice should raise an exception");
+        tr.close();
+    }
+
+    /**
+     * Test behavior if the stream is closed.
+     *
+     * @throws InterruptedException Thread interrupted.
+     * @throws IOException          Error reading or writing mocked data.
+     */
+    @Test
+    void streamClosedTest() throws InterruptedException, IOException {
+        // Callback counter for some unhandy async mockery.
+        final AtomicInteger counter = new AtomicInteger(0);
+
+        // The list which will be populated by the callback.
+        Deque<Trip> trips = new ConcurrentLinkedDeque<>();
+
+        // Start with V1 data and read file to mock list.
+        readLinesToMock(1, "/__files/stream_V1_stops_all.txt", 8);
+
+        AsyncUraTripReader tr = new AsyncUraTripReader(
+                new URL(httpMock.baseUrl() + "/interfaces/ura/stream_V1"),
+                Collections.singletonList(
+                        trip -> {
+                            trips.add(trip);
+                            counter.incrementAndGet();
+                        }
+                )
+        );
+
+        // Open the reader.
+        tr.open();
+
+        // Read for 100ms.
+        TimeUnit.MILLISECONDS.sleep(100);
+        assumeTrue(trips.isEmpty(), "Trips should empty after 100ms without reading");
+
+        // Wait for 1s for the callback to be triggered.
+        TimeUnit.SECONDS.sleep(1);
+
+        assumeTrue(1 == trips.size(), "Unexpected number of trips after first entry");
+
+        // Close the stream.
+        tr.close();
+
+        // Wait for another second.
+        TimeUnit.MILLISECONDS.sleep(100);
+        assertThat("Unexpected number of trips after all lines have been flushed", trips.size(), is(1));
+
+    }
+
+    /**
+     * Read an input file to the line buffer.
+     *
+     * @param version      API version.
+     * @param resourceFile Resource file name.
+     * @param chunks       Number of chunks.
+     */
+    private void readLinesToMock(int version, String resourceFile, int chunks) {
+        WireMock.stubFor(get(urlPathEqualTo("/interfaces/ura/stream_V" + version))
+                .willReturn(aResponse()
+                        .withTransformer("stream-transformer", "source", resourceFile)
+                        .withTransformer("stream-transformer", "chunks", chunks)
+                )
+        );
+    }
+
+    public static class StreamTransformer extends ResponseTransformer {
+        @Override
+        public Response transform(Request request, Response response, FileSource files, Parameters parameters) {
+            int chunks = parameters.getInt("chunks", 1);
+            return Response.Builder.like(response)
+                    // Read source file to response.
+                    .body(() -> AsyncUraTripReaderTest.class.getResourceAsStream(parameters.getString("source")))
+                    // Split response in given number of chunks with 500ms delay.
+                    .chunkedDribbleDelay(new ChunkedDribbleDelay(chunks, chunks * 500))
+                    .build();
+        }
+
+        @Override
+        public String getName() {
+            return "stream-transformer";
+        }
+    }
+}

src/test/resources/__files/instant_V1_messages.txt

@@ -0,0 +1,3 @@
+[4,"1.0",1572882473479]
+[2,"Berensberger Str.","100707","",0,50.8087069,6.0607177,"016e1231d4e30014_100707",0,3,"Sehr geehrte Fahrgäste, wegen Strassenbauarbeiten kann diese Haltestelle nicht von den Bussen der Linien 17, 44 und N2 angefahren werden."]
+[2,"Herzogenr. Rathaus","210511","",0,50.8718175,6.1025675,"016e2cc3a3750006_210511",0,0,"Sehr geehrte Fahrgäste, diese Haltestelle wird vorübergehend von den Linien 47, 147 und N3 nicht angefahren."]

src/test/resources/__files/instant_V1_trips_stop_name.txt

@@ -0,0 +1,11 @@
+[4,"2.0",1483362959788]
+[1,"Uniklinik","100600","H.2",0,50.7756388,6.04425,11,"33","33","1","Aachen Fuchserde","Aachen Fuchserde","318",92000043013001,1483362935000]
+[1,"Uniklinik","100600","H.1",0,50.7756388,6.04425,1,"5","5","1","Driescher Hof-Brand","Driescher Hof-Brand","312",92000282009001,1483362936000]
+[1,"Uniklinik","100600","H.4",0,50.7756388,6.04425,33,"45","45","1","Uniklinik","Uniklinik","317",92000285009001,1483363294000]
+[1,"Uniklinik","100600","H.3",0,50.7756388,6.04425,28,"10","3.B","1","Uniklinik-Ponttor-Hbf.","Uniklinik-Ponttor-Hbf.","347",92000053015001,1483363039000]
+[1,"Uniklinik","100600","H.3",0,50.7756388,6.04425,29,"33","33","1","Uniklinik","Uniklinik","529",92000209014001,1483363288000]
+[1,"Uniklinik","100600","H.2",0,50.7756388,6.04425,1,"73","73","1","Aachen Bf.Rothe Erde","Aachen Bf.Rothe Erde","315",92000291016001,1483363080000]
+[1,"Uniklinik","100600","H.3",0,50.7756388,6.04425,29,"10","3.B","1","Uniklinik-Ponttor-Hbf.","Uniklinik-Ponttor-Hbf.","347",92000053015001,1483363099000]
+[1,"Uniklinik","100600","H.1",0,50.7756388,6.04425,28,"3","3.A","1","Uniklinik-Schanz-Hbf.","Uniklinik-Schanz-Hbf.","325",92000288012001,1483363080000]
+[1,"Uniklinik","100600","H.2",0,50.7756388,6.04425,6,"70","70","1","Aachen Adenauerallee","Aachen Adenauerallee","588",92000225009001,1483363346000]
+[1,"Uniklinik","100600","H.1",0,50.7756388,6.04425,1,"3","3.A","1","Schanz-Hbf.-Ponttor","Schanz-Hbf.-Ponttor",null,92000288013001,1483363380000]

src/test/resources/__files/instant_V2_messages_stop.txt

@@ -0,0 +1,2 @@
+[4,"2.0",1572882473479]
+[2,"Berensberger Str.","100707","",0,50.8087069,6.0607177,"016e1231d4e30014_100707",0,3,"Sehr geehrte Fahrgäste, wegen Strassenbauarbeiten kann diese Haltestelle nicht von den Bussen der Linien 17, 44 und N2 angefahren werden."]

src/test/resources/__files/stream_V1_stops_all.txt

@@ -0,0 +1,8 @@
+[4,"1.0",1542370828725]
+[1,"Campus Melaten","100629","",0,50.78247,6.05053,4,"764","3B","2","Ponttor-Hbf.-Schanz","Ponttor-Hbf.-Schanz","327001",16000304013001,1542375720000]
+[1,"Eckenerstraße","100308","",0,50.7539658,6.1541161,15,"65","65","2","Elisenbrunnen","Elisenbrunnen","301001",16000428004001,1542372660000]
+[1,"Talbot","100111","",0,50.7845802,6.1093236,51,"1","1","2","Süsterau-Uniklinik","Süsterau-Uniklinik","305001",16000351007001,1542372900000]
+[1,"Herz. Schulzentrum","210541","",0,50.8642111,6.1053944,1,"831","HZ1","1","Hofstadt","Hofstadt","737001",16000212021001,1542375000000]
+[1,"Weisweiler Ziegelei","213237","",0,50.8254738,6.325058,14,"96","96","1","Langerwehe Schulzentr.","Langerwehe Schulzentr.",null,16000417012001,1542373320000]
+[1,"Pongs","100233","",0,50.7725688,6.1285466,24,"7","7","2","Eilendorf Am Tunnel","Eilendorf Am Tunnel","540001",16000444014001,1542377460000]
+[1,"Velau","215624","",0,50.7893811,6.2223038,17,"8","8","2","Stolberg Mühlener Bf.","Stolberg Mühlener Bf.","568001",16000319014001,1542374400000]

src/test/resources/__files/stream_V2_stops_all.txt

@@ -0,0 +1,8 @@
+[4,"2.0",1542370788379]
+[1,"Campus Melaten","100629","",0,50.78247,6.05053,4,"764","3B","2","Ponttor-Hbf.-Schanz","Ponttor-Hbf.-Schanz","327001","16000304013001",1542375720000]
+[1,"Eckenerstraße","100308","",0,50.7539658,6.1541161,15,"65","65","2","Elisenbrunnen","Elisenbrunnen","301001","16000428004001",1542372660000]
+[1,"Talbot","100111","",0,50.7845802,6.1093236,51,"1","1","2","Süsterau-Uniklinik","Süsterau-Uniklinik","305001","16000351007001",1542372900000]
+[1,"Herz. Schulzentrum","210541","",0,50.8642111,6.1053944,1,"831","HZ1","1","Hofstadt","Hofstadt","737001","16000212021001",1542375000000]
+[1,"Weisweiler Ziegelei","213237","",0,50.8254738,6.325058,14,"96","96","1","Langerwehe Schulzentr.","Langerwehe Schulzentr.",null,"16000417012001",1542373320000]
+[1,"Pongs","100233","",0,50.7725688,6.1285466,24,"7","7","2","Eilendorf Am Tunnel","Eilendorf Am Tunnel","540001","16000444014001",1542377460000]
+[1,"Velau","215624","",0,50.7893811,6.2223038,17,"8","8","2","Stolberg Mühlener Bf.","Stolberg Mühlener Bf.","568001","16000319014001",1542374400000]

src/test/resources/de/stklcode/pubtrans/ura/instant_V1_trips_stop_name.txt

@@ -1,11 +0,0 @@
-[4,"2.0",1483362959788]
-[1,"Uniklinik","100600","H.2",0,50.7756388,6.04425,11,"33","33",1,"Aachen Fuchserde","Aachen Fuchserde","318",92000043013001,1483362935000]
-[1,"Uniklinik","100600","H.1",0,50.7756388,6.04425,1,"5","5",1,"Driescher Hof-Brand","Driescher Hof-Brand","312",92000282009001,1483362936000]
-[1,"Uniklinik","100600","H.4",0,50.7756388,6.04425,33,"45","45",1,"Uniklinik","Uniklinik","317",92000285009001,1483363294000]
-[1,"Uniklinik","100600","H.3",0,50.7756388,6.04425,28,"10","3.B",1,"Uniklinik-Ponttor-Hbf.","Uniklinik-Ponttor-Hbf.","347",92000053015001,1483363039000]
-[1,"Uniklinik","100600","H.3",0,50.7756388,6.04425,29,"33","33",1,"Uniklinik","Uniklinik","529",92000209014001,1483363288000]
-[1,"Uniklinik","100600","H.2",0,50.7756388,6.04425,1,"73","73",1,"Aachen Bf.Rothe Erde","Aachen Bf.Rothe Erde","315",92000291016001,1483363080000]
-[1,"Uniklinik","100600","H.3",0,50.7756388,6.04425,29,"10","3.B",1,"Uniklinik-Ponttor-Hbf.","Uniklinik-Ponttor-Hbf.","347",92000053015001,1483363099000]
-[1,"Uniklinik","100600","H.1",0,50.7756388,6.04425,28,"3","3.A",1,"Uniklinik-Schanz-Hbf.","Uniklinik-Schanz-Hbf.","325",92000288012001,1483363080000]
-[1,"Uniklinik","100600","H.2",0,50.7756388,6.04425,6,"70","70",1,"Aachen Adenauerallee","Aachen Adenauerallee","588",92000225009001,1483363346000]
-[1,"Uniklinik","100600","H.1",0,50.7756388,6.04425,1,"3","3.A",1,"Schanz-Hbf.-Ponttor","Schanz-Hbf.-Ponttor","325",92000288013001,1483363380000]