2 * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
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
10 package org.opendaylight.netconf.test.tool;
12 import com.ning.http.client.AsyncCompletionHandler;
13 import com.ning.http.client.AsyncHttpClient;
14 import com.ning.http.client.AsyncHttpClientConfig;
15 import com.ning.http.client.HttpResponseStatus;
16 import com.ning.http.client.ListenableFuture;
17 import com.ning.http.client.Realm;
18 import com.ning.http.client.Request;
19 import com.ning.http.client.Response;
20 import java.io.IOException;
21 import java.util.ArrayList;
22 import java.util.concurrent.Callable;
23 import java.util.concurrent.ExecutionException;
24 import java.util.concurrent.Semaphore;
25 import org.slf4j.Logger;
26 import org.slf4j.LoggerFactory;
28 public class Execution implements Callable<Void> {
30 private final ArrayList<Request> payloads;
31 private final AsyncHttpClient asyncHttpClient;
32 private static final Logger LOG = LoggerFactory.getLogger(Execution.class);
33 private final boolean invokeAsync;
34 private final Semaphore semaphore;
35 private final int throttle;
37 static final class DestToPayload {
39 private final String destination;
40 private final String payload;
42 DestToPayload(String destination, String payload) {
43 this.destination = destination;
44 this.payload = payload;
47 public String getDestination() {
51 public String getPayload() {
56 public Execution(TesttoolParameters params, ArrayList<DestToPayload> payloads) {
57 this.invokeAsync = params.async;
58 this.throttle = params.throttle / params.threadAmount;
60 if (params.async && params.threadAmount > 1) {
61 LOG.info("Throttling per thread: {}", this.throttle);
63 this.semaphore = new Semaphore(this.throttle);
65 this.asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder()
66 .setConnectTimeout(Integer.MAX_VALUE)
67 .setRequestTimeout(Integer.MAX_VALUE)
68 .setAllowPoolingConnections(true)
71 this.payloads = new ArrayList<>();
72 for (DestToPayload payload : payloads) {
73 AsyncHttpClient.BoundRequestBuilder requestBuilder = asyncHttpClient.preparePost(payload.getDestination())
74 .addHeader("Content-Type", "application/json")
75 .addHeader("Accept", "application/json")
76 .setBody(payload.getPayload())
77 .setRequestTimeout(Integer.MAX_VALUE);
79 if (params.auth != null) {
80 requestBuilder.setRealm(new Realm.RealmBuilder()
81 .setScheme(Realm.AuthScheme.BASIC)
82 .setPrincipal(params.auth.get(0))
83 .setPassword(params.auth.get(1))
84 .setMethodName("POST")
85 .setUsePreemptiveAuth(true)
88 this.payloads.add(requestBuilder.build());
92 private void invokeSync() {
93 LOG.info("Begin sending sync requests");
94 for (Request request : payloads) {
96 Response response = asyncHttpClient.executeRequest(request).get();
97 if (response.getStatusCode() != 200 && response.getStatusCode() != 204) {
98 if (response.getStatusCode() == 409) {
99 LOG.warn("Request failed, status code: {} - one or more of the devices"
100 + " is already configured, skipping the whole batch", response.getStatusCode());
102 LOG.warn("Status code: {}", response.getStatusCode());
103 LOG.warn("url: {}", request.getUrl());
104 LOG.warn(response.getResponseBody());
107 } catch (InterruptedException | ExecutionException | IOException e) {
108 LOG.warn(e.toString());
111 LOG.info("End sending sync requests");
114 private void invokeAsync() {
115 final ArrayList<ListenableFuture<Response>> futures = new ArrayList<>();
116 LOG.info("Begin sending async requests");
118 for (final Request request : payloads) {
121 } catch (InterruptedException e) {
122 LOG.warn("Semaphore acquire interrupted");
124 futures.add(asyncHttpClient.executeRequest(request, new AsyncCompletionHandler<Response>() {
126 public STATE onStatusReceived(HttpResponseStatus status) throws Exception {
127 super.onStatusReceived(status);
128 if (status.getStatusCode() != 200 && status.getStatusCode() != 204) {
129 if (status.getStatusCode() == 409) {
130 LOG.warn("Request failed, status code: {} - one or more of the devices"
131 + " is already configured, skipping the whole batch", status.getStatusCode());
133 LOG.warn("Request failed, status code: {}",
134 status.getStatusCode() + status.getStatusText());
135 LOG.warn("request: {}", request.toString());
138 return STATE.CONTINUE;
142 public Response onCompleted(Response response) throws Exception {
148 LOG.info("Requests sent, waiting for responses");
151 semaphore.acquire(this.throttle);
152 } catch (InterruptedException e) {
153 LOG.warn("Semaphore acquire interrupted");
156 LOG.info("Responses received, ending...");
160 public Void call() throws Exception {