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.Realm;
17 import com.ning.http.client.Request;
18 import com.ning.http.client.Response;
19 import java.io.IOException;
20 import java.util.ArrayList;
21 import java.util.concurrent.Callable;
22 import java.util.concurrent.ExecutionException;
23 import java.util.concurrent.Semaphore;
24 import org.slf4j.Logger;
25 import org.slf4j.LoggerFactory;
27 public class Execution implements Callable<Void> {
29 private final ArrayList<Request> payloads;
30 private final AsyncHttpClient asyncHttpClient;
31 private static final Logger LOG = LoggerFactory.getLogger(Execution.class);
32 private final boolean invokeAsync;
33 private final Semaphore semaphore;
34 private final int throttle;
36 static final class DestToPayload {
38 private final String destination;
39 private final String payload;
41 DestToPayload(String destination, String payload) {
42 this.destination = destination;
43 this.payload = payload;
46 public String getDestination() {
50 public String getPayload() {
55 public Execution(TesttoolParameters params, ArrayList<DestToPayload> payloads) {
56 this.invokeAsync = params.async;
57 this.throttle = params.throttle / params.threadAmount;
59 if (params.async && params.threadAmount > 1) {
60 LOG.info("Throttling per thread: {}", this.throttle);
62 this.semaphore = new Semaphore(this.throttle);
64 this.asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder()
65 .setConnectTimeout(Integer.MAX_VALUE)
66 .setRequestTimeout(Integer.MAX_VALUE)
67 .setAllowPoolingConnections(true)
70 this.payloads = new ArrayList<>();
71 for (DestToPayload payload : payloads) {
72 AsyncHttpClient.BoundRequestBuilder requestBuilder = asyncHttpClient.preparePost(payload.getDestination())
73 .addHeader("Content-Type", "application/json")
74 .addHeader("Accept", "application/json")
75 .setBody(payload.getPayload())
76 .setRequestTimeout(Integer.MAX_VALUE);
78 if (params.auth != null) {
79 requestBuilder.setRealm(new Realm.RealmBuilder()
80 .setScheme(Realm.AuthScheme.BASIC)
81 .setPrincipal(params.auth.get(0))
82 .setPassword(params.auth.get(1))
83 .setMethodName("POST")
84 .setUsePreemptiveAuth(true)
87 this.payloads.add(requestBuilder.build());
91 private void invokeSync() {
92 LOG.info("Begin sending sync requests");
93 for (Request request : payloads) {
95 Response response = asyncHttpClient.executeRequest(request).get();
96 if (response.getStatusCode() != 200 && response.getStatusCode() != 204) {
97 if (response.getStatusCode() == 409) {
98 LOG.warn("Request failed, status code: {} - one or more of the devices"
99 + " is already configured, skipping the whole batch", response.getStatusCode());
101 LOG.warn("Status code: {}", response.getStatusCode());
102 LOG.warn("url: {}", request.getUrl());
103 LOG.warn(response.getResponseBody());
106 } catch (InterruptedException | ExecutionException | IOException e) {
107 LOG.warn(e.toString());
110 LOG.info("End sending sync requests");
113 private void invokeAsync() {
114 LOG.info("Begin sending async requests");
116 for (final Request request : payloads) {
119 } catch (InterruptedException e) {
120 LOG.warn("Semaphore acquire interrupted");
122 asyncHttpClient.executeRequest(request, new AsyncCompletionHandler<Response>() {
124 public STATE onStatusReceived(HttpResponseStatus status) throws Exception {
125 super.onStatusReceived(status);
126 if (status.getStatusCode() != 200 && status.getStatusCode() != 204) {
127 if (status.getStatusCode() == 409) {
128 LOG.warn("Request failed, status code: {} - one or more of the devices"
129 + " is already configured, skipping the whole batch", status.getStatusCode());
131 LOG.warn("Request failed, status code: {}",
132 status.getStatusCode() + status.getStatusText());
133 LOG.warn("request: {}", request.toString());
136 return STATE.CONTINUE;
140 public Response onCompleted(Response response) {
146 LOG.info("Requests sent, waiting for responses");
149 semaphore.acquire(this.throttle);
150 } catch (InterruptedException e) {
151 LOG.warn("Semaphore acquire interrupted");
154 LOG.info("Responses received, ending...");