netconf/tools/netconf-testtool/src/main/java/org/opendaylight/netconf/test/tool/Execution.java

   1 /*
   2  * Copyright (c) 2016 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
   9
  10 package org.opendaylight.netconf.test.tool;
  11
  12 import com.ning.http.client.*;
  13 import org.slf4j.Logger;
  14 import org.slf4j.LoggerFactory;
  15
  16 import java.io.IOException;
  17 import java.util.ArrayList;
  18 import java.util.concurrent.Callable;
  19 import java.util.concurrent.ExecutionException;
  20 import java.util.concurrent.Semaphore;
  21
  22 public class Execution implements Callable<Void> {
  23
  24     private final ArrayList<Request> payloads;
  25     private final AsyncHttpClient asyncHttpClient;
  26     private static final Logger LOG = LoggerFactory.getLogger(Execution.class);
  27     private final boolean invokeAsync;
  28     private final Semaphore semaphore;
  29     private final int throttle;
  30
  31     static final class DestToPayload {
  32
  33         private final String destination;
  34         private final String payload;
  35
  36         public DestToPayload(String destination, String payload) {
  37             this.destination = destination;
  38             this.payload = payload;
  39         }
  40
  41         public String getDestination() {
  42             return destination;
  43         }
  44
  45         public String getPayload() {
  46             return payload;
  47         }
  48     }
  49
  50     public Execution(TesttoolParameters params, ArrayList<DestToPayload> payloads) {
  51         this.invokeAsync = params.async;
  52         this.throttle = params.throttle / params.threadAmount;
  53
  54         if (params.async && params.threadAmount > 1) {
  55             LOG.info("Throttling per thread: {}", this.throttle);
  56         }
  57         this.semaphore = new Semaphore(this.throttle);
  58
  59         this.asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder()
  60                 .setConnectTimeout(Integer.MAX_VALUE)
  61                 .setRequestTimeout(Integer.MAX_VALUE)
  62                 .setAllowPoolingConnections(true)
  63                 .build());
  64
  65         this.payloads = new ArrayList<>();
  66         for (DestToPayload payload : payloads) {
  67             AsyncHttpClient.BoundRequestBuilder requestBuilder = asyncHttpClient.preparePut(payload.getDestination())
  68                     .addHeader("Content-Type", "application/xml")
  69                     .addHeader("Accept", "application/xml")
  70                     .setBody(payload.getPayload())
  71                     .setRequestTimeout(Integer.MAX_VALUE);
  72
  73             if (params.auth != null) {
  74                 requestBuilder.setRealm(new Realm.RealmBuilder()
  75                         .setScheme(Realm.AuthScheme.BASIC)
  76                         .setPrincipal(params.auth.get(0))
  77                         .setPassword(params.auth.get(1))
  78                         .setMethodName("PUT")
  79                         .setUsePreemptiveAuth(true)
  80                         .build());
  81             }
  82             this.payloads.add(requestBuilder.build());
  83         }
  84     }
  85
  86     private void invokeSync() {
  87         LOG.info("Begin sending sync requests");
  88         for (Request request : payloads) {
  89             try {
  90                 Response response = asyncHttpClient.executeRequest(request).get();
  91                 if (response.getStatusCode() != 200 && response.getStatusCode() != 204) {
  92                     LOG.warn("Status code: {}", response.getStatusCode());
  93                     LOG.warn("url: {}", request.getUrl());
  94                     LOG.warn(response.getResponseBody());
  95                 }
  96             } catch (InterruptedException | ExecutionException | IOException e) {
  97                 LOG.warn(e.toString());
  98             }
  99         }
 100         LOG.info("End sending sync requests");
 101     }
 102
 103     private void invokeAsync() {
 104         final ArrayList<ListenableFuture<Response>> futures = new ArrayList<>();
 105         LOG.info("Begin sending async requests");
 106
 107         for (final Request request : payloads) {
 108             try {
 109                 semaphore.acquire();
 110             } catch (InterruptedException e) {
 111                 LOG.warn("Semaphore acquire interrupted");
 112             }
 113             futures.add(asyncHttpClient.executeRequest(request, new AsyncCompletionHandler<Response>() {
 114                 @Override
 115                 public STATE onStatusReceived(HttpResponseStatus status) throws Exception {
 116                     super.onStatusReceived(status);
 117                     if (status.getStatusCode() != 200 && status.getStatusCode() != 204) {
 118                         LOG.warn("Request failed, status code: {}", status.getStatusCode() + status.getStatusText());
 119                         LOG.warn("request: {}", request.toString());
 120                     }
 121                     return STATE.CONTINUE;
 122                 }
 123
 124                 @Override
 125                 public Response onCompleted(Response response) throws Exception {
 126                     semaphore.release();
 127                     return response;
 128                 }
 129             }));
 130         }
 131         LOG.info("Requests sent, waiting for responses");
 132
 133         try {
 134             semaphore.acquire(this.throttle);
 135         } catch (InterruptedException e) {
 136             LOG.warn("Semaphore acquire interrupted");
 137         }
 138
 139         LOG.info("Responses received, ending...");
 140     }
 141
 142     @Override
 143     public Void call() throws Exception {
 144         if (invokeAsync) {
 145             this.invokeAsync();
 146         } else {
 147             this.invokeSync();
 148         }
 149         return null;
 150     }
 151 }