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
8 package org.opendaylight.netconf.test.tool;
10 import com.ning.http.client.AsyncCompletionHandler;
11 import com.ning.http.client.AsyncHttpClient;
12 import com.ning.http.client.AsyncHttpClientConfig;
13 import com.ning.http.client.HttpResponseStatus;
14 import com.ning.http.client.Realm;
15 import com.ning.http.client.Request;
16 import com.ning.http.client.Response;
17 import java.io.IOException;
18 import java.util.ArrayList;
19 import java.util.concurrent.Callable;
20 import java.util.concurrent.ExecutionException;
21 import java.util.concurrent.Semaphore;
22 import org.slf4j.Logger;
23 import org.slf4j.LoggerFactory;
25 public class Execution implements Callable<Void> {
27 private final ArrayList<Request> payloads;
28 private final AsyncHttpClient asyncHttpClient;
29 private static final Logger LOG = LoggerFactory.getLogger(Execution.class);
30 private final boolean invokeAsync;
31 private final Semaphore semaphore;
32 private final int throttle;
34 static final class DestToPayload {
36 private final String destination;
37 private final String payload;
39 DestToPayload(final String destination, final String payload) {
40 this.destination = destination;
41 this.payload = payload;
44 public String getDestination() {
48 public String getPayload() {
53 public Execution(final TesttoolParameters params, final ArrayList<DestToPayload> payloads) {
54 this.invokeAsync = params.async;
55 this.throttle = params.throttle / params.threadAmount;
57 if (params.async && params.threadAmount > 1) {
58 LOG.info("Throttling per thread: {}", this.throttle);
60 this.semaphore = new Semaphore(this.throttle);
62 this.asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder()
63 .setConnectTimeout(Integer.MAX_VALUE)
64 .setRequestTimeout(Integer.MAX_VALUE)
65 .setAllowPoolingConnections(true)
68 this.payloads = new ArrayList<>();
69 for (DestToPayload payload : payloads) {
70 AsyncHttpClient.BoundRequestBuilder requestBuilder = asyncHttpClient.preparePost(payload.getDestination())
71 .addHeader("Content-Type", "application/json")
72 .addHeader("Accept", "application/json")
73 .setBody(payload.getPayload())
74 .setRequestTimeout(Integer.MAX_VALUE);
76 requestBuilder.setRealm(new Realm.RealmBuilder()
77 .setScheme(Realm.AuthScheme.BASIC)
78 .setPrincipal(params.controllerAuthUsername)
79 .setPassword(params.controllerAuthPassword)
80 .setMethodName("POST")
81 .setUsePreemptiveAuth(true)
84 this.payloads.add(requestBuilder.build());
88 private void invokeSync() {
89 LOG.info("Begin sending sync requests");
90 for (Request request : payloads) {
92 Response response = asyncHttpClient.executeRequest(request).get();
93 if (response.getStatusCode() != 200 && response.getStatusCode() != 204) {
94 if (response.getStatusCode() == 409) {
95 LOG.warn("Request failed, status code: {} - one or more of the devices"
96 + " is already configured, skipping the whole batch", response.getStatusCode());
98 LOG.warn("Status code: {}", response.getStatusCode());
99 LOG.warn("url: {}", request.getUrl());
100 LOG.warn("body: {}", response.getResponseBody());
103 } catch (InterruptedException | ExecutionException | IOException e) {
104 LOG.warn("Failed to execute request", e);
107 LOG.info("End sending sync requests");
110 private void invokeAsync() {
111 LOG.info("Begin sending async requests");
113 for (final Request request : payloads) {
116 } catch (InterruptedException e) {
117 LOG.warn("Semaphore acquire interrupted");
119 asyncHttpClient.executeRequest(request, new AsyncCompletionHandler<Response>() {
121 public STATE onStatusReceived(final HttpResponseStatus status) throws Exception {
122 super.onStatusReceived(status);
123 if (status.getStatusCode() != 200 && status.getStatusCode() != 204) {
124 if (status.getStatusCode() == 409) {
125 LOG.warn("Request failed, status code: {} - one or more of the devices"
126 + " is already configured, skipping the whole batch", status.getStatusCode());
128 LOG.warn("Request failed, status code: {}",
129 status.getStatusCode() + status.getStatusText());
130 LOG.warn("request: {}", request.toString());
133 return STATE.CONTINUE;
137 public Response onCompleted(final Response response) {
143 LOG.info("Requests sent, waiting for responses");
146 semaphore.acquire(this.throttle);
147 } catch (InterruptedException e) {
148 LOG.warn("Semaphore acquire interrupted");
151 LOG.info("Responses received, ending...");