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.preparePost(payload.getDestination())
  68                     .addHeader("Content-Type", "application/json")
  69                     .addHeader("Accept", "application/json")
  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("POST")
  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                     if (response.getStatusCode() == 409) {
  93                         LOG.warn("Request failed, status code: {} - one or more of the devices" +
  94                                 " is already configured, skipping the whole batch", response.getStatusCode());
  95                     } else {
  96                         LOG.warn("Status code: {}", response.getStatusCode());
  97                         LOG.warn("url: {}", request.getUrl());
  98                         LOG.warn(response.getResponseBody());
  99                     }
 100                 }
 101             } catch (InterruptedException | ExecutionException | IOException e) {
 102                 LOG.warn(e.toString());
 103             }
 104         }
 105         LOG.info("End sending sync requests");
 106     }
 107
 108     private void invokeAsync() {
 109         final ArrayList<ListenableFuture<Response>> futures = new ArrayList<>();
 110         LOG.info("Begin sending async requests");
 111
 112         for (final Request request : payloads) {
 113             try {
 114                 semaphore.acquire();
 115             } catch (InterruptedException e) {
 116                 LOG.warn("Semaphore acquire interrupted");
 117             }
 118             futures.add(asyncHttpClient.executeRequest(request, new AsyncCompletionHandler<Response>() {
 119                 @Override
 120                 public STATE onStatusReceived(HttpResponseStatus status) throws Exception {
 121                     super.onStatusReceived(status);
 122                     if (status.getStatusCode() != 200 && status.getStatusCode() != 204) {
 123                         if (status.getStatusCode() == 409) {
 124                             LOG.warn("Request failed, status code: {} - one or more of the devices" +
 125                                     " is already configured, skipping the whole batch", status.getStatusCode());
 126                         } else {
 127                             LOG.warn("Request failed, status code: {}", status.getStatusCode() + status.getStatusText());
 128                             LOG.warn("request: {}", request.toString());
 129                         }
 130                     }
 131                     return STATE.CONTINUE;
 132                 }
 133
 134                 @Override
 135                 public Response onCompleted(Response response) throws Exception {
 136                     semaphore.release();
 137                     return response;
 138                 }
 139             }));
 140         }
 141         LOG.info("Requests sent, waiting for responses");
 142
 143         try {
 144             semaphore.acquire(this.throttle);
 145         } catch (InterruptedException e) {
 146             LOG.warn("Semaphore acquire interrupted");
 147         }
 148
 149         LOG.info("Responses received, ending...");
 150     }
 151
 152     @Override
 153     public Void call() throws Exception {
 154         if (invokeAsync) {
 155             this.invokeAsync();
 156         } else {
 157             this.invokeSync();
 158         }
 159         return null;
 160     }
 161 }