restconf/restconf-client-impl/src/main/java/org/opendaylight/yangtools/restconf/client/RestListenableEventStreamContext.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.yangtools.restconf.client;
   9
  10 import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
  11
  12 import java.io.UnsupportedEncodingException;
  13 import java.lang.reflect.InvocationTargetException;
  14 import java.lang.reflect.Method;
  15 import java.net.URI;
  16 import java.net.URLEncoder;
  17 import java.util.Date;
  18 import java.util.concurrent.Callable;
  19 import java.util.concurrent.ExecutionException;
  20 import java.util.concurrent.Executors;
  21
  22 import javax.ws.rs.core.MediaType;
  23
  24 import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
  25 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  26 import org.opendaylight.yangtools.restconf.client.api.event.EventStreamInfo;
  27 import org.opendaylight.yangtools.restconf.client.api.event.EventStreamReplay;
  28 import org.opendaylight.yangtools.restconf.client.api.event.ListenableEventStreamContext;
  29 import org.opendaylight.yangtools.restconf.client.to.RestRpcResult;
  30 import org.opendaylight.yangtools.restconf.common.ResourceUri;
  31 import org.opendaylight.yangtools.websocket.client.WebSocketIClient;
  32 import org.opendaylight.yangtools.websocket.client.callback.ClientMessageCallback;
  33 import org.opendaylight.yangtools.yang.binding.NotificationListener;
  34 import org.opendaylight.yangtools.yang.binding.util.BindingReflections;
  35 import org.opendaylight.yangtools.yang.common.RpcResult;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 import com.google.common.base.Charsets;
  40 import com.google.common.base.Function;
  41 import com.google.common.base.Optional;
  42 import com.google.common.util.concurrent.ListenableFuture;
  43 import com.google.common.util.concurrent.ListeningExecutorService;
  44 import com.google.common.util.concurrent.MoreExecutors;
  45 import com.sun.jersey.api.client.ClientResponse;
  46
  47
  48
  49
  50 public class RestListenableEventStreamContext<L extends NotificationListener> implements ListenableEventStreamContext,ClientMessageCallback {
  51
  52     private static final Logger logger = LoggerFactory.getLogger(RestListenableEventStreamContext.class.toString());
  53     private final ListeningExecutorService pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(10));
  54     private WebSocketIClient wsClient;
  55     private Method listenerCallbackMethod;
  56     private final RestconfClientImpl restconfClient;
  57     private final EventStreamInfo streamInfo;
  58
  59     public RestListenableEventStreamContext(final EventStreamInfo streamInfo,final RestconfClientImpl restconfClient){
  60         this.restconfClient = restconfClient;
  61         this.streamInfo = streamInfo;
  62     }
  63
  64     @Override
  65     public <T extends NotificationListener> ListenerRegistration<T> registerNotificationListener(final T listener) {
  66
  67         for (Method m:listener.getClass().getDeclaredMethods()){
  68             if (BindingReflections.isNotificationCallback(m)){
  69                 this.listenerCallbackMethod = m;
  70                 break;
  71             }
  72         }
  73         return new AbstractListenerRegistration<T>(listener) {
  74             @Override
  75             protected void removeRegistration() {
  76                 stopListening();
  77             }
  78         };
  79     }
  80
  81     @Override
  82     public ListenableFuture<RpcResult<Void>> startListening() {
  83
  84
  85         ClientResponse response = null;
  86         try {
  87             response = extractWebSocketUriFromRpc(this.streamInfo.getIdentifier());
  88         } catch (ExecutionException e) {
  89             logger.trace("Execution exception while extracting stream name {}",e);
  90             throw new IllegalStateException(e);
  91         } catch (InterruptedException e) {
  92             logger.trace("InterruptedException while extracting stream name {}",e);
  93             throw new IllegalStateException(e);
  94         } catch (UnsupportedEncodingException e) {
  95             logger.trace("UnsupportedEncodingException while extracting stream name {}",e);
  96             throw new IllegalStateException(e);
  97         }
  98         boolean success = true;
  99         if (response.getStatus() != 200) {
 100             success = false;
 101         }
 102
 103         final RestRpcResult rpcResult = new RestRpcResult(success,response.getLocation());
 104         createWebsocketClient(response.getLocation());
 105
 106         ListenableFuture<RpcResult<Void>> future = pool.submit(new Callable<RpcResult<Void>>() {
 107             @Override
 108             public RpcResult<Void> call() {
 109                 return rpcResult;
 110             }
 111         });
 112
 113         return future;
 114     }
 115
 116     @Override
 117     public ListenableFuture<RpcResult<Void>> startListeningWithReplay(final Optional<Date> startTime, final Optional<Date> endTime) {
 118         //TODO RESTCONF doesn't provide this functionality
 119         return null;
 120     }
 121
 122     @Override
 123     public void stopListening() {
 124         this.wsClient.writeAndFlush(new CloseWebSocketFrame(42,this.streamInfo.getIdentifier()));
 125     }
 126
 127     @Override
 128     public ListenableFuture<Optional<EventStreamReplay>> getReplay(final Optional<Date> startTime, final Optional<Date> endTime) {
 129         //TODO RESTCONF doesn't provide this functionality
 130         return null;
 131     }
 132
 133     @Override
 134     public void close() {
 135         this.stopListening();
 136     }
 137
 138     private ClientResponse extractWebSocketUriFromRpc(final String methodName) throws ExecutionException, InterruptedException, UnsupportedEncodingException {
 139         ListenableFuture<ClientResponse> clientFuture = restconfClient.get(ResourceUri.STREAM.getPath()+"/"+encodeUri(this.streamInfo.getIdentifier()),MediaType.APPLICATION_XML,new Function<ClientResponse, ClientResponse>(){
 140
 141             @Override
 142             public ClientResponse apply(final ClientResponse clientResponse) {
 143                 return clientResponse;
 144             }
 145         });
 146
 147         return clientFuture.get();
 148     }
 149     private void createWebsocketClient(final URI websocketServerUri){
 150         this.wsClient = new WebSocketIClient(websocketServerUri,this);
 151     }
 152     private String encodeUri(final String encodedPart) throws UnsupportedEncodingException {
 153         return URI.create(URLEncoder.encode(encodedPart, Charsets.US_ASCII.name()).toString()).toASCIIString();
 154     }
 155
 156     @Override
 157     public void onMessageReceived(final Object message) {
 158         if (null == this.listenerCallbackMethod){
 159             throw new IllegalStateException("No listener method to invoke.");
 160         }
 161         try {
 162             this.listenerCallbackMethod.invoke(message);
 163         } catch (IllegalAccessException | InvocationTargetException e) {
 164             throw new IllegalStateException("Failed to invoke callback", e);
 165         }
 166     }
 167
 168 }