Merge branch 'release-0.6.3'

README.md

@@ -14,7 +14,7 @@ repositories {
 }
 
 dependencies {
- compile group: 'org.radarcns', name: 'radar-commons', version: '0.6.2'
+ compile group: 'org.radarcns', name: 'radar-commons', version: '0.6.3'
 }
 ```
 
@@ -26,7 +26,7 @@ repositories {
 }
 
 dependencies {
- testCompile group: 'org.radarcns', name: 'radar-commons-testing', version: '0.6.2'
+ testCompile group: 'org.radarcns', name: 'radar-commons-testing', version: '0.6.3'
 }
 ```
 
@@ -51,7 +51,7 @@ configurations.all {
 }
 
 dependencies {
- compile group: 'org.radarcns', name: 'radar-commons', version: '0.6.3-SNAPSHOT', changing: true
+ compile group: 'org.radarcns', name: 'radar-commons', version: '0.6.4-SNAPSHOT', changing: true
 }
 ```
 

build.gradle

@@ -36,7 +36,7 @@ allprojects {
  // Configuration //
  //---------------------------------------------------------------------------//
 
- version = '0.6.2'
+ version = '0.6.3'
  group = 'org.radarcns'
  ext.githubRepoName = 'RADAR-CNS/RADAR-Commons'
 
@@ -234,6 +234,7 @@ dependencies {
  testImplementation group: 'org.mockito', name: 'mockito-core', version: mockitoVersion
  testImplementation group: 'org.hamcrest', name: 'hamcrest-all', version: hamcrestVersion
  testImplementation group: 'com.squareup.okhttp3', name: 'mockwebserver', version: okhttpVersion
+ testRuntime group: 'org.slf4j', name: 'slf4j-simple', version: slf4jVersion
 
  codacy group: 'com.github.codacy', name: 'codacy-coverage-reporter', version: codacyVersion
 }

src/main/java/org/radarcns/producer/BatchedKafkaSender.java

@@ -51,12 +51,12 @@ public <L extends K, W extends V> KafkaTopicSender<L, W> sender(final AvroTopic<
  }
 
  @Override
- public boolean isConnected() {
+ public boolean isConnected() throws AuthenticationException {
  return wrappedSender.isConnected();
  }
 
  @Override
- public boolean resetConnection() {
+ public boolean resetConnection() throws AuthenticationException {
  return wrappedSender.resetConnection();
  }
 

src/main/java/org/radarcns/producer/KafkaSender.java

@@ -33,10 +33,10 @@ <L extends K, W extends V> KafkaTopicSender<L, W> sender(AvroTopic<L, W> topic)
  * If the sender is no longer connected, try to reconnect.
  * @return whether the connection has been restored.
  */
- boolean resetConnection();
+ boolean resetConnection() throws AuthenticationException;
 
  /**
  * Whether the sender is connected to the Kafka system.
  */
- boolean isConnected();
+ boolean isConnected() throws AuthenticationException;
 }

src/main/java/org/radarcns/producer/ThreadedKafkaSender.java

@@ -70,12 +70,17 @@ public Thread newThread( Runnable r) {
  public void run() {
  opsRequests.add(1);
 
- boolean success = sendHeartbeat();
- if (success) {
- state.didConnect();
- } else {
- logger.error("Failed to send message");
- state.didDisconnect();
+ try {
+ boolean success = sendHeartbeat();
+ if (success) {
+ state.didConnect();
+ } else {
+ logger.error("Failed to send message");
+ state.didDisconnect();
+ }
+ } catch (AuthenticationException ex) {
+ logger.error("Unauthorized");
+ state.wasUnauthorized();
  }
 
  if (opsSent.hasAverage() && opsRequests.hasAverage()) {
@@ -196,6 +201,10 @@ public void run() {
 
  if (exception == null) {
  state.didConnect();
+ } else if (exception instanceof AuthenticationException) {
+ logger.error("Authentication failed");
+ state.wasUnauthorized();
+ break;
  } else {
  logger.error("Failed to send message");
  state.didDisconnect();
@@ -207,7 +216,7 @@ public void run() {
  }
  }
 
- private boolean sendHeartbeat() {
+ private boolean sendHeartbeat() throws AuthenticationException {
  boolean success = false;
  for (int i = 0; !success && i < RETRIES; i++) {
  success = wrappedSender.resetConnection();
@@ -216,12 +225,14 @@ private boolean sendHeartbeat() {
  }
 
  @Override
- public synchronized boolean isConnected() {
+ public synchronized boolean isConnected() throws AuthenticationException {
  switch (state.getState()) {
  case CONNECTED:
  return true;
  case DISCONNECTED:
  return false;
+ case UNAUTHORIZED:
+ throw new AuthenticationException("Authorization invalid");
  case UNKNOWN:
  state.didDisconnect();
  return false;
@@ -231,15 +242,22 @@ public synchronized boolean isConnected() {
  }
 
  @Override
- public boolean resetConnection() {
+ public boolean resetConnection() throws AuthenticationException {
  if (isConnected()) {
  return true;
- } else if (wrappedSender.resetConnection()) {
- state.didConnect();
- return true;
- } else {
- state.didDisconnect();
- return false;
+ }
+
+ try {
+ if (wrappedSender.resetConnection()) {
+ state.didConnect();
+ return true;
+ } else {
+ state.didDisconnect();
+ return false;
+ }
+ } catch (AuthenticationException ex) {
+ state.wasUnauthorized();
+ throw ex;
  }
  }
 

src/main/java/org/radarcns/producer/rest/ConnectionState.java

@@ -34,7 +34,7 @@ public final class ConnectionState {
 
  /** State symbols of the connection. */
  public enum State {
- CONNECTED, DISCONNECTED, UNKNOWN
+ CONNECTED, DISCONNECTED, UNKNOWN, UNAUTHORIZED
  }
 
  private long timeout;
@@ -72,6 +72,14 @@ public synchronized void didDisconnect() {
  state = State.DISCONNECTED;
  }
 
+ public synchronized void wasUnauthorized() {
+ state = State.UNAUTHORIZED;
+ }
+
+ public synchronized void reset() {
+ state = State.UNKNOWN;
+ }
+
  /**
  * Set the timeout after which the state will go from CONNECTED to UNKNOWN.
  * @param timeout timeout

src/main/java/org/radarcns/producer/rest/RestSender.java

@@ -139,6 +139,7 @@ private void setRestClient(RestClient newClient) {
  throw new IllegalArgumentException("Schemaless topics do not have a valid URL", ex);
  }
  httpClient = newClient;
+ state.reset();
  }
 
  public final synchronized void setSchemaRetriever(SchemaRetriever retriever) {
@@ -179,6 +180,7 @@ public synchronized Headers getHeaders() {
 
  public synchronized void setHeaders(Headers additionalHeaders) {
  this.additionalHeaders = additionalHeaders;
+ this.state.reset();
  }
 
  private class RestTopicSender<L extends K, W extends V> implements KafkaTopicSender<L, W> {
@@ -365,35 +367,42 @@ public <L extends K, W extends V> KafkaTopicSender<L, W> sender(AvroTopic<L, W>
  }
 
  @Override
- public boolean resetConnection() {
+ public boolean resetConnection() throws AuthenticationException {
  if (state.getState() == State.CONNECTED) {
  return true;
  }
  try (Response response = httpClient.request(getIsConnectedRequest())) {
  if (response.isSuccessful()) {
  state.didConnect();
- return true;
+ } else if (response.code() == 401) {
+ state.wasUnauthorized();
  } else {
  state.didDisconnect();
  String bodyString = responseBody(response);
  logger.warn("Failed to make heartbeat request to {} (HTTP status code {}): {}",
  httpClient, response.code(), bodyString);
- return false;
  }
  } catch (IOException ex) {
  // no stack trace is needed
  state.didDisconnect();
  logger.warn("Failed to make heartbeat request to {}: {}", httpClient, ex.toString());
- return false;
  }
+
+ if (state.getState() == State.UNAUTHORIZED) {
+ throw new AuthenticationException("HEAD request unauthorized");
+ }
+
+ return state.getState() == State.CONNECTED;
  }
 
- public boolean isConnected() {
+ public boolean isConnected() throws AuthenticationException {
  switch (state.getState()) {
  case CONNECTED:
  return true;
  case DISCONNECTED:
  return false;
+ case UNAUTHORIZED:
+ throw new AuthenticationException("Unauthorized");
  case UNKNOWN:
  return resetConnection();
  default:

src/test/java/org/radarcns/producer/rest/RestSenderTest.java

@@ -34,6 +34,7 @@
 import org.radarcns.data.SpecificRecordEncoder;
 import org.radarcns.kafka.ObservationKey;
 import org.radarcns.passive.phone.PhoneLight;
+import org.radarcns.producer.AuthenticationException;
 import org.radarcns.producer.KafkaTopicSender;
 import org.radarcns.topic.AvroTopic;
 
@@ -187,6 +188,38 @@ public void resetConnection() throws Exception {
  assertEquals("HEAD", request.getMethod());
  }
 
+ @Test
+ public void resetConnectionUnauthorized() throws Exception {
+ webServer.enqueue(new MockResponse().setResponseCode(401));
+ try {
+ sender.isConnected();
+ fail("Authentication exception expected");
+ } catch (AuthenticationException ex) {
+ // success
+ }
+ try {
+ sender.isConnected();
+ fail("Authentication exception expected");
+ } catch (AuthenticationException ex) {
+ // success
+ }
+ webServer.enqueue(new MockResponse().setResponseCode(401));
+ try {
+ sender.resetConnection();
+ fail("Authentication exception expected");
+ } catch (AuthenticationException ex) {
+ assertEquals(2, webServer.getRequestCount());
+ // success
+ }
+ webServer.enqueue(new MockResponse().setResponseCode(200));
+ try {
+ assertTrue(sender.resetConnection());
+ } catch (AuthenticationException ex) {
+ assertEquals(3, webServer.getRequestCount());
+ fail("Unexpected authentication failure");
+ }
+ }
+
  @Test
  public void withCompression() throws IOException, InterruptedException {
  sender.setCompression(true);