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.*;
13 import org.slf4j.Logger;
14 import org.slf4j.LoggerFactory;
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;
22 public class Execution implements Callable<Void> {
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;
31 static final class DestToPayload {
33 private final String destination;
34 private final String payload;
36 public DestToPayload(String destination, String payload) {
37 this.destination = destination;
38 this.payload = payload;
41 public String getDestination() {
45 public String getPayload() {
50 public Execution(TesttoolParameters params, ArrayList<DestToPayload> payloads) {
51 this.invokeAsync = params.async;
52 this.throttle = params.throttle / params.threadAmount;
54 if (params.async && params.threadAmount > 1) {
55 LOG.info("Throttling per thread: {}", this.throttle);
57 this.semaphore = new Semaphore(this.throttle);
59 this.asyncHttpClient = new AsyncHttpClient(new AsyncHttpClientConfig.Builder()
60 .setConnectTimeout(Integer.MAX_VALUE)
61 .setRequestTimeout(Integer.MAX_VALUE)
62 .setAllowPoolingConnections(true)
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);
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)
82 this.payloads.add(requestBuilder.build());
86 private void invokeSync() {
87 LOG.info("Begin sending sync requests");
88 for (Request request : payloads) {
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());
96 LOG.warn("Status code: {}", response.getStatusCode());
97 LOG.warn("url: {}", request.getUrl());
98 LOG.warn(response.getResponseBody());
101 } catch (InterruptedException | ExecutionException | IOException e) {
102 LOG.warn(e.toString());
105 LOG.info("End sending sync requests");
108 private void invokeAsync() {
109 final ArrayList<ListenableFuture<Response>> futures = new ArrayList<>();
110 LOG.info("Begin sending async requests");
112 for (final Request request : payloads) {
115 } catch (InterruptedException e) {
116 LOG.warn("Semaphore acquire interrupted");
118 futures.add(asyncHttpClient.executeRequest(request, new AsyncCompletionHandler<Response>() {
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());
127 LOG.warn("Request failed, status code: {}", status.getStatusCode() + status.getStatusText());
128 LOG.warn("request: {}", request.toString());
131 return STATE.CONTINUE;
135 public Response onCompleted(Response response) throws Exception {
141 LOG.info("Requests sent, waiting for responses");
144 semaphore.acquire(this.throttle);
145 } catch (InterruptedException e) {
146 LOG.warn("Semaphore acquire interrupted");
149 LOG.info("Responses received, ending...");
153 public Void call() throws Exception {