001/*
002 * Licensed to DuraSpace under one or more contributor license agreements.
003 * See the NOTICE file distributed with this work for additional information
004 * regarding copyright ownership.
005 *
006 * DuraSpace licenses this file to you under the Apache License,
007 * Version 2.0 (the "License"); you may not use this file except in
008 * compliance with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018
019package org.fcrepo.integration.http.api;
020
021import static javax.ws.rs.core.HttpHeaders.LINK;
022import static javax.ws.rs.core.Response.Status.CREATED;
023import static org.junit.Assert.assertEquals;
024
025import java.io.IOException;
026import java.time.Duration;
027import java.util.ArrayList;
028import java.util.Objects;
029import java.util.concurrent.ExecutorService;
030import java.util.concurrent.Executors;
031import java.util.concurrent.Future;
032import java.util.concurrent.Phaser;
033import java.util.concurrent.atomic.AtomicInteger;
034
035import org.apache.http.client.methods.CloseableHttpResponse;
036import org.junit.AfterClass;
037import org.junit.BeforeClass;
038import org.junit.Test;
039import org.slf4j.Logger;
040import org.slf4j.LoggerFactory;
041import org.springframework.test.context.TestExecutionListeners;
042
043import com.google.common.base.Stopwatch;
044
045/**
046 * @author pwinckles
047 */
048@TestExecutionListeners(
049        listeners = { TestIsolationExecutionListener.class },
050        mergeMode = TestExecutionListeners.MergeMode.MERGE_WITH_DEFAULTS)
051public class ConcurrencyIT extends AbstractResourceIT {
052
053    private static final Logger LOGGER = LoggerFactory.getLogger(ConcurrencyIT.class);
054
055    private static final int THREAD_COUNT = 4;
056
057    private static ExecutorService executor;
058
059    @BeforeClass
060    public static void beforeClass() {
061        executor = Executors.newFixedThreadPool(THREAD_COUNT);
062    }
063
064    @AfterClass
065    public static void afterClass() {
066        executor.shutdown();
067    }
068
069    @Test
070    public void basicContainerPosts() {
071        final var phaser = new Phaser(THREAD_COUNT + 1);
072        final var testDuration = Duration.ofSeconds(15);
073        final var tasks = new ArrayList<Future<Void>>(THREAD_COUNT);
074        final var succeeded = new AtomicInteger(0);
075        final var failed = new AtomicInteger(0);
076
077        for (int i = 0; i < THREAD_COUNT; i++) {
078            tasks.add(executor.submit(() -> {
079                phaser.arriveAndAwaitAdvance();
080                final var stopwatch = Stopwatch.createStarted();
081
082                while (stopwatch.elapsed().compareTo(testDuration) < 0) {
083                    if (postCreateContainer()) {
084                        succeeded.incrementAndGet();
085                    } else {
086                        failed.incrementAndGet();
087                    }
088                }
089                return null;
090            }));
091        }
092
093        phaser.arriveAndAwaitAdvance();
094
095        tasks.forEach(future -> {
096            try {
097                future.get();
098            } catch (Exception e) {
099                throw new RuntimeException(e);
100            }
101        });
102
103        final var total = succeeded.get() + failed.get();
104
105        assertEquals(String.format("%s requests out of %s failed", failed.get(), total),
106                0, failed.get());
107    }
108
109    private boolean postCreateContainer() {
110        final var post = postObjMethod();
111        post.setHeader(LINK, BASIC_CONTAINER_LINK_HEADER);
112        try (final CloseableHttpResponse response = execute(post)) {
113            if (Objects.equals(CREATED.getStatusCode(), response.getStatusLine().getStatusCode())) {
114                return true;
115            } else {
116                LOGGER.error("Concurrent request failed: {}", response.getStatusLine());
117            }
118        } catch (IOException e) {
119            LOGGER.error("Failed to execute request", e);
120            return false;
121        }
122        return false;
123    }
124
125}