package org.onap.policy.std;
import java.net.URI;
-import javax.websocket.ClientEndpoint;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.onap.policy.api.NotificationHandler;
import org.onap.policy.common.logging.flexlogger.Logger;
import org.onap.policy.xacml.api.XACMLErrorConstants;
-@ClientEndpoint
public class AutoClientEnd extends WebSocketClient {
private static StdPDPNotification notification = null;
private static StdPDPNotification oldNotification = null;
private static String url = null;
private static boolean status = false;
private static boolean stop = false;
- private static boolean message = false;
private static boolean error = false;
+ private static boolean restartNeeded = false;
+ private static ScheduledExecutorService restartExecutorService = null;
private static Logger logger = FlexLogger.getLogger(AutoClientEnd.class.getName());
private AutoClientEnd(URI serverUri) {
@Override
public void onMessage(String msg) {
logger.info("Received Auto Notification from : " + getURI() + ", Notification: " + msg);
- AutoClientEnd.message = true;
try {
AutoClientEnd.notification = NotificationUnMarshal.notificationJSON(msg);
} catch (Exception e) {
AutoClientEnd.oldNotification = AutoClientEnd.notification;
callHandler();
}
-
- AutoClientEnd.message = false;
}
@Override
public void onClose(int code, String reason, boolean remote) {
logger.info("AutoClientEnd disconnected from: " + getURI() + "; Code: " + code + ", reason : " + reason);
- if (!AutoClientEnd.stop && !AutoClientEnd.message) {
- // This Block of code is executed if there is any Network Failure or
- // if the Notification is Down.
- logger.error(XACMLErrorConstants.ERROR_SYSTEM_ERROR + "Disconnected from Notification Server");
- AutoClientEnd.client = null;
- AutoClientEnd.status = false;
- // Try to connect Back to available PDP.
- AutoClientEnd.error = true;
- start(url);
- }
- AutoClientEnd.message = false;
+ AutoClientEnd.restartNeeded = true;
}
@Override
public void onError(Exception ex) {
logger.error("XACMLErrorConstants.ERROR_PROCESS_FLOW + Error connecting to: " + getURI()
+ ", Exception occured ...\n" + ex);
- // trying to Restart by self.
- stop();
- if (AutoClientEnd.url != null) {
- AutoClientEnd.client = null;
- AutoClientEnd.status = false;
- AutoClientEnd.error = true;
- AutoClientEnd.start(AutoClientEnd.url);
- }
+ AutoClientEnd.restartNeeded = true;
}
@Override
public void onOpen(ServerHandshake arg0) {
+ restartNeeded = false;
logger.info("Auto Notification Session Started... " + getURI());
}
+ private static void restart() {
+ try {
+ if (client != null && restartNeeded && !stop) {
+ logger.info("Auto Notification Session Restarting ... " + getUrl());
+ client.reconnect();
+ }
+ } catch (Exception e) {
+ logger.info("Auto Notification Session Error Started... " + getUrl());
+ }
+ }
+
/**
* Sets the auto.
*
client = new AutoClientEnd(new URI(url + "notifications"));
client.connect();
status = true;
+ restartExecutorService = Executors.newSingleThreadScheduledExecutor();
+ Runnable task = AutoClientEnd::restart;
+ restartExecutorService.scheduleAtFixedRate(task, 60, 60, TimeUnit.SECONDS);
+
if (error) {
// will not trigger. leave it in to later add checks
// The URL's will be in Sync according to design Spec.
}
} catch (Exception e) {
logger.error(XACMLErrorConstants.ERROR_SYSTEM_ERROR + e);
- client = null;
status = false;
changeUrl();
}
private static void changeUrl() {
// Change the PDP if it is not Up.
+ stop();
StdPolicyEngine.rotatePDPList();
start(StdPolicyEngine.getPDPURL());
}
}
logger.info("\n Closing Auto Notification WebSocket Connection.. ");
stop = true;
+ // first stop the restart service
+ try {
+ restartExecutorService.shutdown();
+ } catch (Exception e1) {
+ logger.info("\n AutoClientEnd: Error stoppping the restart Scheduler ");
+ }
+
+ // close the connection
try {
client.closeBlocking();
- } catch (InterruptedException e) {
- logger.info("\n Error Closing Auto Notification WebSocket Connection.. InterruptedException");
+ } catch (Exception e) {
+ logger.error("\n ERROR Closing Auto Notification WebSocket Connection.. ");
}
+
logger.info("\n Closed the Auto Notification WebSocket Connection.. ");
client = null;
status = false;
stop = false;
+ restartNeeded = false;
}
private static void callHandler() {
import java.net.URI;
import java.util.concurrent.CountDownLatch;
-import javax.websocket.ClientEndpoint;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.onap.policy.api.NotificationScheme;
import org.onap.policy.std.StdPDPNotification;
import org.onap.policy.xacml.api.XACMLErrorConstants;
-@ClientEndpoint
public class ManualClientEnd extends WebSocketClient {
private static CountDownLatch latch;
private static StdPDPNotification notification = null;
logger.info("Manual Notification Recieved Message from : " + getURI() + ", Notification: " + message);
ManualClientEnd.resultJson = message;
try {
- ManualClientEnd.notification = NotificationUnMarshal.notificationJSON(message);
- latch.countDown();
+ ManualClientEnd.notification = NotificationUnMarshal.notificationJSON(message);
} catch (Exception e) {
logger.error(XACMLErrorConstants.ERROR_DATA_ISSUE + e);
- latch.countDown();
}
+ latch.countDown();
}
@Override
client = new ManualClientEnd(new URI(url + "notifications"));
client.connect();
latch.await();
- client.close();
+ client.closeBlocking();
} catch (Exception e) {
logger.error(XACMLErrorConstants.ERROR_SYSTEM_ERROR + e);
}
package org.onap.policy.std.test;
import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import org.java_websocket.WebSocket;
import org.java_websocket.handshake.ClientHandshake;
import org.java_websocket.server.WebSocketServer;
import org.onap.policy.api.NotificationScheme;
import org.onap.policy.api.PDPNotification;
import org.onap.policy.std.AutoClientEnd;
-import org.onap.policy.std.StdPDPNotification;
import org.springframework.util.SocketUtils;
/**
public class AutoClientEndTest {
private static WebSocketServer ws;
- private static int port = 18080;
+ private static int port = SocketUtils.findAvailableTcpPort();
private static CountDownLatch countServerDownLatch = null;
- private StdPDPNotification notification = null;
+ private static PDPNotification notification = null;
/**
* Start server.
*/
@BeforeClass
public static void startServer() throws Exception {
- port = SocketUtils.findAvailableTcpPort();
- ws = new WebSocketServer(new InetSocketAddress(port), 16) {
+ notification = null;
+ ws = new WebSocketServer(new InetSocketAddress(port), 1) {
@Override
public void onOpen(WebSocket conn, ClientHandshake handshake) {
conn.send("{\"removedPolicies\": [],\"loadedPolicies\": "
};
- ws.setConnectionLostTimeout(30);
+ ws.setConnectionLostTimeout(0);
ws.start();
}
NotificationHandler handler = new NotificationHandler() {
@Override
- public void notificationReceived(PDPNotification notifi) {
- notification = (StdPDPNotification) notifi;
+ public void notificationReceived(PDPNotification notify) {
+ notification = notify;
countServerDownLatch.countDown();
}
countServerDownLatch = new CountDownLatch(1);
AutoClientEnd.start("http://localhost:" + port + "/");
- countServerDownLatch.await();
+ countServerDownLatch.await(45, TimeUnit.SECONDS);
+
+ assertNotNull(notification);
+
+ // simulate a server restart and verify client reconnects
+ countServerDownLatch = new CountDownLatch(1);
+ ws.stop(30000);
+ startServer();
+ countServerDownLatch.await(60+10, TimeUnit.SECONDS);
assertNotNull(notification);
- assertTrue(AutoClientEnd.getStatus());
+
+ AutoClientEnd.stop();
+
}
@AfterClass
- public static void successTests() throws InterruptedException, IOException {
- AutoClientEnd.stop();
- ws.stop();
+ public static void stopServer() throws InterruptedException, IOException {
+ ws.stop(30000);
}
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import org.java_websocket.WebSocket;
import org.java_websocket.handshake.ClientHandshake;
import org.java_websocket.server.WebSocketServer;
/**
* The class <code>ManualClientEndTest</code> contains tests for the class <code>{@link ManualClientEnd}</code>.
*
- * @generatedBy CodePro at 6/1/16 1:41 PM
- * @version $Revision: 1.0 $
*/
public class ManualClientEndTest {
private static WebSocketServer ws;
- private static int port = 18080;
+ private static int port = SocketUtils.findAvailableTcpPort();
private static CountDownLatch countServerDownLatch = null;
private static String recvMsg = null;
*/
@BeforeClass
public static void startServer() throws Exception {
- port = SocketUtils.findAvailableTcpPort();
- ws = new WebSocketServer(new InetSocketAddress(port), 16) {
+ ws = new WebSocketServer(new InetSocketAddress(port), 1) {
@Override
public void onOpen(WebSocket conn, ClientHandshake handshake) {
};
- ws.setConnectionLostTimeout(30);
+ ws.setConnectionLostTimeout(0);
ws.start();
}
@Test
- public void testAutoClient() throws Exception {
+ public void testManualClient() throws Exception {
countServerDownLatch = new CountDownLatch(1);
ManualClientEnd.start("http://localhost:" + port + "/");
- countServerDownLatch.await();
+ countServerDownLatch.await(45, TimeUnit.SECONDS);
assertNotNull(ManualClientEnd.result(NotificationScheme.MANUAL_ALL_NOTIFICATIONS));
assertTrue("Manual".equalsIgnoreCase(recvMsg));
@AfterClass
public static void successTests() throws InterruptedException, IOException {
- ws.stop();
+ ws.stop(30000);
}
}
Code Review / policy / engine.git / commitdiff