Java 类org.apache.commons.lang3.concurrent.ConcurrentUtils 实例源码

@Override
public <T> Future<T> execute(final Controller controller, final BackgroundActionRegistry registry, final BackgroundAction<T> action) {
    if(log.isDebugEnabled()) {
        log.debug(String.format("Run action %s in background", action));
    }
    registry.add(action);
    action.init();
    // Start background task
    final Callable<T> command = new BackgroundCallable<T>(action, controller, registry);
    try {
        final Future<T> task = concurrentExecutor.execute(command);
        if(log.isInfoEnabled()) {
            log.info(String.format("Scheduled background runnable %s for execution", action));
        }
        return task;
    }
    catch(RejectedExecutionException e) {
        log.error(String.format("Error scheduling background task %s for execution. %s", action, e.getMessage()));
        action.cleanup();
        return ConcurrentUtils.constantFuture(null);
    }
}

/**
 * Returns thread factory for producing threads associated with the specified
 * group name. The group name-space is hierarchical, based on slash-delimited
 * name segments, e.g. {@code onos/intent}.
 *
 * @param groupName group name
 * @return thread factory
 */
public static GroupedThreadFactory groupedThreadFactory(String groupName) {
    GroupedThreadFactory factory = FACTORIES.get(groupName);
    if (factory != null) {
        return factory;
    }

    // Find the parent group or root the group hierarchy under default group.
    int i = groupName.lastIndexOf(DELIMITER);
    if (i > 0) {
        String name = groupName.substring(0, i);
        ThreadGroup parentGroup = groupedThreadFactory(name).threadGroup();
        factory = new GroupedThreadFactory(new ThreadGroup(parentGroup, groupName));
    } else {
        factory = new GroupedThreadFactory(new ThreadGroup(groupName));
    }

    return ConcurrentUtils.putIfAbsent(FACTORIES, groupName, factory);
}

@Before
public void setUp() {
    config = Configuration.buildFromConfig("config-mock.properties");
    when(connectionStrategy.getHttpClient()).thenReturn(asyncHttpClient);
    HttpResponse response = mock(HttpResponse.class);
    Future<HttpResponse> future = ConcurrentUtils.constantFuture(response);
    when(asyncHttpClient.execute(any(HttpUriRequest.class),any(FutureCallback.class))).thenReturn(future, null);
    HttpEntity httpEntity = mock(HttpEntity.class);
    when(response.getEntity()).thenReturn(httpEntity);
    StatusLine statusLine = mock(StatusLine.class);
    when(response.getStatusLine()).thenReturn(statusLine);
    when(statusLine.getStatusCode()).thenReturn(200);
    try {
        InputStream inputStream = IOUtils.toInputStream(SERVER_RESPONSE_EXPECTED, "UTF-8");
        when(httpEntity.getContent()).thenReturn(inputStream);
        client = new LogInsightClient(config, connectionStrategy);
        // client.connect(user, password);
        assertEquals("Invalid session id!!",
                "qyOLWEe7f/GjdM1WnczrCeQure97B/NpTbWTeqqYPBd1AYMf9cMNfQYqltITI4ffPMx822Sz9i/X47t8VwsDb0oGckclJUdn83cyIPk6WlsOpI4Yjw6WpurAnv9RhDsYSzKhAMzskzhTOJKfDHZjWR5v576WwtJA71wqI7igFrG91LG5c/3GfzMb68sUHF6hV+meYtGS4A1y/lUItvfkqTTAxBtTCZNoKrvCJZ4R+b6vuAAYoBNSWL7ycIy2LsALrVFxftAkA8n9DBAZYA9T5A==",
                client.getSessionId());
    } catch (Exception e) {
        logger.error("Exception raised " + ExceptionUtils.getStackTrace(e));
    }
}

/**
 * Returns thread factory for producing threads associated with the specified
 * group name. The group name-space is hierarchical, based on slash-delimited
 * name segments, 
 *
 * @param groupName group name
 * @return thread factory
 */
public static AtriumGroupedThreadFactory groupedThreadFactory(String groupName) {
    AtriumGroupedThreadFactory factory = FACTORIES.get(groupName);
    if (factory != null) {
        return factory;
    }

    // Find the parent group or root the group hierarchy under default group.
    int i = groupName.lastIndexOf(DELIMITER);
    if (i > 0) {
        String name = groupName.substring(0, i);
        ThreadGroup parentGroup = groupedThreadFactory(name).threadGroup();
        factory = new AtriumGroupedThreadFactory(new ThreadGroup(parentGroup, groupName));
    } else {
        factory = new AtriumGroupedThreadFactory(new ThreadGroup(groupName));
    }

    return ConcurrentUtils.putIfAbsent(FACTORIES, groupName, factory);
}

/**
 * Returns thread factory for producing threads associated with the specified
 * group name. The group name-space is hierarchical, based on slash-delimited
 * name segments, e.g. {@code onos/intent}.
 *
 * @param groupName group name
 * @return thread factory
 */
public static GroupedThreadFactory groupedThreadFactory(String groupName) {
    GroupedThreadFactory factory = FACTORIES.get(groupName);
    if (factory != null) {
        return factory;
    }

    // Find the parent group or root the group hierarchy under default group.
    int i = groupName.lastIndexOf(DELIMITER);
    if (i > 0) {
        String name = groupName.substring(0, i);
        ThreadGroup parentGroup = groupedThreadFactory(name).threadGroup();
        factory = new GroupedThreadFactory(new ThreadGroup(parentGroup, groupName));
    } else {
        factory = new GroupedThreadFactory(new ThreadGroup(groupName));
    }

    return ConcurrentUtils.putIfAbsent(FACTORIES, groupName, factory);
}

/**
 * Reads LogValue from LogMap.
 * <p>
 * It will use the cached value if it exist, if not it will
 * get the value from distributed store.
 *
 * @param key log sequence number
 * @return Future containing log value
 */
public Future<LogValue> getLogValue(final SeqNum key) {
    try {
        return cache.get(key, new Callable<Future<LogValue>>() {

            @Override
            public Future<LogValue> call() throws Exception {
                return logMap.getAsync(key);
            }
        });
    } catch (ExecutionException e) {
        log.error("Reading from Log Map failed.", e);
        // should never happen?
        return ConcurrentUtils.constantFuture(null);
    }
}

/**
 * Publishes ADD Port Event.
 *
 * @param portData the port event to publish
 */
private void publishAddPortEvent(PortData portData) {
    if (!registryService.hasControl(portData.getOriginDpid().value())) {
        log.debug("Not the master for switch {}. Suppressed port add event {}.",
                  portData.getOriginDpid(), portData);
        return;
    }

    // Publish the information
    TopologyBatchOperation tbo = new TopologyBatchOperation();
    TopologyEvent topologyEvent =
        new TopologyEvent(portData, getOnosInstanceId());
    tbo.appendAddOperation(topologyEvent);
    publishTopologyOperations(tbo);

    // Store the new Port Event in the local cache
    ConcurrentMap<ByteBuffer, PortData> portDataEntries =
        ConcurrentUtils.putIfAbsent(publishedPortDataEntries,
                    portData.getDpid(),
                    new ConcurrentHashMap<ByteBuffer, PortData>());
    portDataEntries.put(portData.getIDasByteBuffer(), portData);
}

/**
 * Publishes ADD Link Event.
 *
 * @param linkData the link event to publish
 */
private void publishAddLinkEvent(LinkData linkData) {
    if (!registryService.hasControl(linkData.getOriginDpid().value())) {
        log.debug("Not the master for dst switch {}. Suppressed link add event {}.",
                  linkData.getOriginDpid(), linkData);
        return;
    }

    // Publish the information
    TopologyBatchOperation tbo = new TopologyBatchOperation();
    TopologyEvent topologyEvent =
        new TopologyEvent(linkData, getOnosInstanceId());
    tbo.appendAddOperation(topologyEvent);
    publishTopologyOperations(tbo);

    // Store the new Link Event in the local cache
    ConcurrentMap<ByteBuffer, LinkData> linkDataEntries =
        ConcurrentUtils.putIfAbsent(publishedLinkDataEntries,
                    linkData.getDst().getDpid(),
                    new ConcurrentHashMap<ByteBuffer, LinkData>());
    linkDataEntries.put(linkData.getIDasByteBuffer(), linkData);
}

/**
 * Returns thread factory for producing threads associated with the specified
 * group name. The group name-space is hierarchical, based on slash-delimited
 * name segments, e.g. {@code onos/intent}.
 *
 * @param groupName group name
 * @return thread factory
 */
public static GroupedThreadFactory groupedThreadFactory(String groupName) {
    GroupedThreadFactory factory = FACTORIES.get(groupName);
    if (factory != null) {
        return factory;
    }

    // Find the parent group or root the group hierarchy under default group.
    int i = groupName.lastIndexOf(DELIMITER);
    if (i > 0) {
        String name = groupName.substring(0, i);
        ThreadGroup parentGroup = groupedThreadFactory(name).threadGroup();
        factory = new GroupedThreadFactory(new ThreadGroup(parentGroup, groupName));
    } else {
        factory = new GroupedThreadFactory(new ThreadGroup(groupName));
    }

    return ConcurrentUtils.putIfAbsent(FACTORIES, groupName, factory);
}

public Future<Boolean> run(String...args) {
    String[] argA = new String[2 + args.length];
    argA[0] = DEFAULT_SHELL;
    argA[1] = scriptFile;
    for (int i = 0; i < args.length; i++) {
        argA[2+i] = args[i];
    }
    ProcessBuilder scriptPB = new ProcessBuilder(argA);
    scriptPB.redirectErrorStream(true);

    try {
        Process scriptProcess = scriptPB.start();
        Future<Boolean> scriptStatus = TimerManager.get().submit( new TrackShell(scriptProcess) );
        return scriptStatus;
    } catch (IOException e) {
        log.error("Error invoking script: " + scriptFile, e);
        return ConcurrentUtils.constantFuture(false);
    }
}

/**
 * Will queue up the <code>BackgroundAction</code> to be run in a background thread
 *
 * @param action The runnable to execute in a secondary thread
 */
@Override
public <T> Future<T> background(final BackgroundAction<T> action) {
    if(registry.contains(action)) {
        log.warn(String.format("Skip duplicate background action %s found in registry", action));
        return ConcurrentUtils.constantFuture(null);
    }
    return DefaultBackgroundExecutor.get().execute(this, registry, action);
}

@Override
public Future<Response> getJsonObject(
        SuccessCallback success,
        HttpErrorCallback error,
        FailureCallback failure,
        String resourcePath
) {
    LOG.info("Invoking test druid webservice: {}", this);

    // Store the most recent query sent through
    lastUrl = resourcePath;

    // Invoke failure callback if we have a throwable to give it
    if (throwable != null) {
        failure.invoke(throwable);
        return CompletedFuture.throwing(throwable);
    }

     try {
         if (statusCode == 200) {
            success.invoke(mapper.readTree(jsonResponse.call()));
        } else {
            error.invoke(statusCode, reasonPhrase, jsonResponse.call());
        }
    } catch (IOException e) {
        failure.invoke(e);
         return CompletedFuture.throwing(throwable);
    }

    return ConcurrentUtils.constantFuture(null);
}

@SuppressWarnings("unchecked")
@Override
public <T> FutureObject<T> saveAsync(Object object) {
    ObjectHolder objectHolder = new ObjectHolder(object);

    setIdIfNecessary(objectHolder);

    MockStore.put(objectHolder.getId(), object, tx());

    Future<?> futureId = ConcurrentUtils.constantFuture(objectHolder.getId());
    return new FutureObject<T>(r, (Future<IdRef<?>>) futureId, (T) object);
}

/**
 * Gets the LogMapManager for given {@link SharedLogObjectID}.
 * <p/>
 * If listener was not registered, it will create and register a listener.
 *
 * @param oid {@link SharedLogObjectID}
 * @return {@link LogMapManager}
 */
private LogMapManager getLogMapManager(final SharedLogObjectID oid) {
    LogMapManager listener
        = ConcurrentUtils.createIfAbsentUnchecked(listenerMap, oid,
            new ConcurrentInitializer<LogMapManager>() {
                @Override
                public LogMapManager get() throws ConcurrentException {
                    IMap<SeqNum, LogValue> logMap = getLogMap(oid);
                    return new LogMapManager(oid, logMap);
                }
            });
    return listener;
}

@Override
public void entryAdded(EntryEvent<SeqNum, LogValue> event) {
    // Cache maintenance
    cache.put(event.getKey(),
              ConcurrentUtils.constantFuture(event.getValue()));

    // TODO will need suppress mechanism once we have health check

    if (lastLog.compareTo(event.getKey()) < 0) {
        lastLog = event.getKey();
    }
    for (LogEventListener lsnr : listeners) {
        lsnr.logAdded(event.getKey());
    }
}

@Override
public void entryUpdated(EntryEvent<SeqNum, LogValue> event) {
    // Cache maintenance
    cache.put(event.getKey(),
              ConcurrentUtils.constantFuture(event.getValue()));

    // only add will be notified to listeners
}

/**
 * Publishes ADD Host Event.
 *
 * @param hostData the host event to publish
 */
private void publishAddHostEvent(HostData hostData) {
    //
    // NOTE: The implementation below assumes that there is just one
    // attachment point stored in hostData. Currently, this assumption
    // is true based on the existing implementation of the caller
    // hostAdded().
    //

    if (!registryService.hasControl(hostData.getOriginDpid().value())) {
        log.debug("Not the master for attachment switch {}. Suppressed host add event {}.",
                  hostData.getOriginDpid(), hostData);
        return;
    }

    // Publish the information
    TopologyBatchOperation tbo = new TopologyBatchOperation();
    TopologyEvent topologyEvent =
        new TopologyEvent(hostData, getOnosInstanceId());
    tbo.appendAddOperation(topologyEvent);
    publishTopologyOperations(tbo);

    // Store the new Host Event in the local cache
    ConcurrentMap<ByteBuffer, HostData> hostDataEntries =
        ConcurrentUtils.putIfAbsent(publishedHostDataEntries,
            hostData.getOriginDpid(),
            new ConcurrentHashMap<ByteBuffer, HostData>());
    hostDataEntries.put(hostData.getIDasByteBuffer(), hostData);
}

@Test
public void testLogEventWithTwoParams() {
    final List<Runnable> runnableList = new ArrayList<>();
    final MutableInt mutableInt = new MutableInt(0);
    doAnswer(new Answer() {
        @Override
        public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
            mutableInt.increment();
            Object[] arguments = invocationOnMock.getArguments();
            runnableList.add((Runnable) arguments[0]);
            if (mutableInt.getValue() == 1) {
                return ConcurrentUtils.constantFuture("aaa");
            } else if (mutableInt.getValue() == 2) {
                return ConcurrentUtils.constantFuture("bbb");
            } else {
                return null;
            }
        }
    }).when(singleThreadExecutor).submit(Mockito.any(Runnable.class));

    // Tested calls
    loggingService.logEvent(Level.INFO, "aaa");
    loggingService.logEvent(Level.ERROR, "bbb");

    assertEquals(2, runnableList.size());

    assertTrue(runnableList.get(0) instanceof LoggingServiceImpl.LoggingDetailsRunnable);
    assertTrue(runnableList.get(1) instanceof LoggingServiceImpl.LoggingDetailsRunnable);

    assertEquals(new LoggingServiceImpl.LoggingDetailsRunnable(Level.INFO, "aaa"), runnableList.get(0));
    assertEquals(new LoggingServiceImpl.LoggingDetailsRunnable(Level.ERROR, "bbb"), runnableList.get(1));
}

@Test
public void testConsolePrint() throws Exception {
    final List<Runnable> runnableList = new ArrayList<>();
    final MutableInt mutableInt = new MutableInt(0);
    doAnswer(new Answer() {
        @Override
        public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
            mutableInt.increment();
            Object[] arguments = invocationOnMock.getArguments();
            runnableList.add((Runnable) arguments[0]);
            if (mutableInt.getValue() == 1) {
                return ConcurrentUtils.constantFuture("firstMessage");
            } else if (mutableInt.getValue() == 2) {
                return ConcurrentUtils.constantFuture("secondMessage");
            } else {
                return null;
            }
        }
    }).when(singleThreadExecutor).submit(Mockito.any(Runnable.class));

    consolePrinter.printWithColor(GREEN, "firstMessage");
    Future lastFuture = consolePrinter.printWithColor(GREEN, "secondMessage");

    assertEquals("secondMessage", lastFuture.get());

    assertEquals(2, runnableList.size());

    assertTrue(runnableList.get(0) instanceof ConsolePrinterImpl.ConsolePrinterRunnable);
    assertTrue(runnableList.get(1) instanceof ConsolePrinterImpl.ConsolePrinterRunnable);

    assertEquals(new ConsolePrinterImpl.ConsolePrinterRunnable(GREEN, "firstMessage"), runnableList.get(0));
    assertEquals(new ConsolePrinterImpl.ConsolePrinterRunnable(GREEN, "secondMessage"), runnableList.get(1));
}

public Future<TransferStatus> submit(final TransferCallable runnable) throws BackgroundException {
    return ConcurrentUtils.constantFuture(runnable.call());
}

/**
 * If TestDruidWebService#throwable is set, invokes the failure callback.  Otherwise invokes success or failure
 * dependent on whether TestDruidWebService#statusCode equals 200.
 * Note that since this doesn't send requests to druid all the responses will be null or {@link CompletedFuture}.
 */
@Override
@SuppressWarnings("checkstyle:cyclomaticcomplexity")
public Future<Response> postDruidQuery(
        RequestContext context,
        SuccessCallback success,
        HttpErrorCallback error,
        FailureCallback failure,
        DruidQuery<?> query
) {
    LOG.info("Invoking test druid webservice: {}", this);

    // Store the most recent query sent through
    lastQuery = query;

    // Convert the query to json
    try {
        jsonQuery = writer.withDefaultPrettyPrinter().writeValueAsString(query);
    } catch (JsonProcessingException ignored) {
        // Ignore
    }

    // Invoke failure callback if we have a throwable to give it
    if (throwable != null) {
        failure.invoke(throwable);
        return CompletedFuture.throwing(throwable);
    }

    if (lastQuery.getQueryType() instanceof DefaultQueryType) {
        // For known response types, create a default response provider

        // Set the response to use based on the type of the query we're processing
        DefaultQueryType defaultQueryType = (DefaultQueryType) lastQuery.getQueryType();
        switch (defaultQueryType) {
            case GROUP_BY:
            case TOP_N:
            case TIMESERIES:
            case LOOKBACK:
                // default response is groupBy response
                break;
            case SEGMENT_METADATA:
                jsonResponse = () -> segmentMetadataResponse;
                break;
            case TIME_BOUNDARY:
                jsonResponse = () -> timeBoundaryResponse;
                break;
            default:
                throw new IllegalArgumentException("Illegal query type : " + lastQuery.getQueryType());
        }

    } else {
        // Otherwise extended query types will have to set up their own responses
    }

    try {
        if (query instanceof WeightEvaluationQuery) {
            success.invoke(mapper.readTree(weightResponse));
        } else if (statusCode == 200) {
            success.invoke(mapper.readTree(jsonResponse.call()));
        } else {
            error.invoke(statusCode, reasonPhrase, jsonResponse.call());
        }
    } catch (IOException e) {
        failure.invoke(e);
        return CompletedFuture.throwing(e);
    }

    return ConcurrentUtils.constantFuture(null);
}

@Override
public Future<RecordMetadata> send(ProducerRecord record) {
  // Fake result: only for testing purpose
  return ConcurrentUtils.constantFuture(new RecordMetadata(null, 0, 0, 0, 0, 0, 0));
}

@Override
public <T> FutureObject<T> fetchAsync(IdRef<T> id) {
    T object = fetch(id);
    Future<?> futureObject = ConcurrentUtils.constantFuture(object);
    return new FutureObject<>(r, futureObject);
}

@Override
public FutureObject<Void> destroyAsync(IdRef<?> id) {
    destroy(id);
    Future<Void> future = ConcurrentUtils.constantFuture(null);
    return new FutureObject<>(r, future);
}

@SuppressWarnings("unchecked")
private <T> FutureObject<T> saveEntityAsync(ObjectHolder objectHolder, Entity entity) {
    Key key = datastore.put(entity);
    Future<?> futureId = ConcurrentUtils.constantFuture(IdRefToKey.toIdRef(r, key, objectHolder.getModel()));
    return new FutureObject<>(r, (Future<IdRef<?>>) futureId, (T) objectHolder.getObject());
}

@Override
public <T> FutureObject<T> fetchAsync(IdRef<T> id) {
    T object = fetch(id);
    Future<?> futureObject = ConcurrentUtils.constantFuture(object);
    return new FutureObject<T>(r, futureObject);
}

@Override
public FutureObject<Void> destroyAsync(IdRef<?> id) {
    destroy(id);
    Future<Void> future = ConcurrentUtils.constantFuture(null);
    return new FutureObject<Void>(r, future);
}

/**
 * Capture arguments to test for expected values.
 * Since there is no response, the future will hold a null value.
 *
 * @param success  callback for handling successful requests.
 * @param error  callback for handling http errors.
 * @param failure  callback for handling exception failures.
 * @param requestBuilder  The bound request builder for the request to be sent.
 * @param timerName  The name that distinguishes this request as part of a druid query or segment metadata request
 * @param outstanding  The counter that keeps track of the outstanding (in flight) requests for the top level query
 *
 * @return a future with a null response.
 */
@Override
protected Future<Response> sendRequest(
        final SuccessCallback success,
        final HttpErrorCallback error,
        final FailureCallback failure,
        final BoundRequestBuilder requestBuilder,
        final String timerName,
        final AtomicLong outstanding
) {
    this.request = requestBuilder.build();
    return ConcurrentUtils.constantFuture(null);
}

private Future<GetItemResult> getNullItemResult() {

        GetItemResult getItemResult = new GetItemResult();

        return ConcurrentUtils.constantFuture(getItemResult);
    }

private Future<SendMessageResult> getSendMessageResult() {

        SendMessageResult sendMessageResult = new SendMessageResult();

        return ConcurrentUtils.constantFuture(sendMessageResult);
    }

/**
 * Map from a keyID to a key. This will be called when decrypting. The data
 * to decrypt will include the keyID used to encrypt it.
 * 
 * @param keyId
 *            The KeyID to map to a key
 */
@Override
public Future<IKey> resolveKeyAsync(String keyId) {
    return ConcurrentUtils.constantFuture(this.keys.get(keyId));
}