| /** |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version |
| * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions |
| * and limitations under the License. |
| */ |
| |
| package org.apache.storm; |
| |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.function.Predicate; |
| import java.util.stream.Collectors; |
| import org.apache.storm.cluster.IStormClusterState; |
| import org.apache.storm.generated.Bolt; |
| import org.apache.storm.generated.GlobalStreamId; |
| import org.apache.storm.generated.Grouping; |
| import org.apache.storm.generated.KillOptions; |
| import org.apache.storm.generated.SpoutSpec; |
| import org.apache.storm.generated.StormTopology; |
| import org.apache.storm.generated.StreamInfo; |
| import org.apache.storm.scheduler.INimbus; |
| import org.apache.storm.task.TopologyContext; |
| import org.apache.storm.testing.CompletableSpout; |
| import org.apache.storm.testing.CompleteTopologyParam; |
| import org.apache.storm.testing.FixedTuple; |
| import org.apache.storm.testing.FixedTupleSpout; |
| import org.apache.storm.testing.MkClusterParam; |
| import org.apache.storm.testing.MkTupleParam; |
| import org.apache.storm.testing.MockedSources; |
| import org.apache.storm.testing.TestJob; |
| import org.apache.storm.testing.TrackedTopology; |
| import org.apache.storm.testing.TupleCaptureBolt; |
| import org.apache.storm.thrift.TException; |
| import org.apache.storm.tuple.Fields; |
| import org.apache.storm.tuple.Tuple; |
| import org.apache.storm.tuple.TupleImpl; |
| import org.apache.storm.utils.ConfigUtils; |
| import org.apache.storm.utils.RegisteredGlobalState; |
| import org.apache.storm.utils.Time; |
| import org.apache.storm.utils.Time.SimulatedTime; |
| import org.apache.storm.utils.Utils; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * A utility that helps with testing topologies, Bolts and Spouts. |
| */ |
| public class Testing { |
| /** |
| * The default amount of wall time should be spent waiting for |
| * specific conditions to happen. Default is 10 seconds unless |
| * the environment variable STORM_TEST_TIMEOUT_MS is set. |
| */ |
| public static final int TEST_TIMEOUT_MS; |
| private static final Logger LOG = LoggerFactory.getLogger(Testing.class); |
| |
| static { |
| int timeout = 10_000; |
| try { |
| timeout = Integer.parseInt(System.getenv("STORM_TEST_TIMEOUT_MS")); |
| } catch (Exception e) { |
| //Ignored, will go with default timeout |
| } |
| TEST_TIMEOUT_MS = timeout; |
| } |
| |
| /** |
| * Continue to execute body repeatedly until condition is true or TEST_TIMEOUT_MS has |
| * passed. |
| * @param condition what we are waiting for |
| * @param body what to run in the loop |
| * @throws AssertionError if the loop timed out. |
| */ |
| public static void whileTimeout(Condition condition, Runnable body) { |
| whileTimeout(TEST_TIMEOUT_MS, condition, body); |
| } |
| |
| /** |
| * Continue to execute body repeatedly until condition is true or TEST_TIMEOUT_MS has |
| * passed. |
| * @param timeoutMs the number of ms to wait before timing out. |
| * @param condition what we are waiting for |
| * @param body what to run in the loop |
| * @throws AssertionError if the loop timed out. |
| */ |
| public static void whileTimeout(long timeoutMs, Condition condition, Runnable body) { |
| long endTime = System.currentTimeMillis() + timeoutMs; |
| LOG.debug("Looping until {}", condition); |
| int count = 0; |
| while (condition.exec()) { |
| count++; |
| if (System.currentTimeMillis() > endTime) { |
| LOG.info("Condition {} not met in {} ms after calling {} times", condition, timeoutMs, count); |
| LOG.info(Utils.threadDump()); |
| throw new AssertionError("Test timed out (" + timeoutMs + "ms) " + condition); |
| } |
| body.run(); |
| } |
| LOG.debug("Condition met {}", condition); |
| } |
| |
| /** |
| * Convenience method for data.stream.allMatch(pred). |
| */ |
| public static <T> boolean isEvery(Collection<T> data, Predicate<T> pred) { |
| return data.stream().allMatch(pred); |
| } |
| |
| /** |
| * Run with simulated time. |
| * |
| * @deprecated use ``` |
| * try (Time.SimulatedTime time = new Time.SimulatedTime()) { |
| * ... |
| * } |
| * ``` |
| * @param code what to run |
| */ |
| @Deprecated |
| public static void withSimulatedTime(Runnable code) { |
| try (SimulatedTime st = new SimulatedTime()) { |
| code.run(); |
| } |
| } |
| |
| private static LocalCluster cluster(MkClusterParam param, boolean simulated) throws Exception { |
| return cluster(param, null, simulated); |
| } |
| |
| private static LocalCluster cluster(MkClusterParam param) throws Exception { |
| return cluster(param, null, false); |
| } |
| |
| private static LocalCluster cluster(MkClusterParam param, String id, boolean simulated) throws Exception { |
| Integer supervisors = param.getSupervisors(); |
| if (supervisors == null) { |
| supervisors = 2; |
| } |
| Integer ports = param.getPortsPerSupervisor(); |
| if (ports == null) { |
| ports = 3; |
| } |
| Map<String, Object> conf = param.getDaemonConf(); |
| if (conf == null) { |
| conf = new HashMap<>(); |
| } |
| return new LocalCluster.Builder() |
| .withSupervisors(supervisors) |
| .withPortsPerSupervisor(ports) |
| .withDaemonConf(conf) |
| .withNimbusDaemon(param.isNimbusDaemon()) |
| .withTracked(id) |
| .withSimulatedTime(simulated) |
| .build(); |
| } |
| |
| /** |
| * Run with a local cluster. |
| * |
| * @deprecated use ``` |
| * try (LocalCluster cluster = new LocalCluster()) { |
| * ... |
| * } |
| * ``` |
| * @param code what to run |
| */ |
| @Deprecated |
| public static void withLocalCluster(TestJob code) { |
| withLocalCluster(new MkClusterParam(), code); |
| } |
| |
| /** |
| * Run with a local cluster. |
| * |
| * @deprecated use ``` |
| * try (LocalCluster cluster = new LocalCluster.Builder()....build()) { |
| * ... |
| * } |
| * ``` |
| * @param param configs to set in the cluster |
| * @param code what to run |
| */ |
| @Deprecated |
| public static void withLocalCluster(MkClusterParam param, TestJob code) { |
| try (LocalCluster lc = cluster(param)) { |
| code.run(lc); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| /** |
| * Run with a local cluster. |
| * |
| * @deprecated use ``` |
| * try (LocalCluster cluster = new LocalCluster.Builder()....build()) { |
| * ... |
| * } |
| * ``` |
| * @param clusterConf some configs to set in the cluster |
| */ |
| @Deprecated |
| public static ILocalCluster getLocalCluster(Map<String, Object> clusterConf) { |
| @SuppressWarnings("unchecked") |
| Map<String, Object> conf = (Map<String, Object>) clusterConf.get("daemon-conf"); |
| if (conf == null) { |
| conf = new HashMap<>(); |
| } |
| Number supervisors = (Number) clusterConf.getOrDefault("supervisors", 2); |
| Number ports = (Number) clusterConf.getOrDefault("ports-per-supervisor", 3); |
| INimbus inimbus = (INimbus) clusterConf.get("inimbus"); |
| Number portMin = (Number) clusterConf.getOrDefault("supervisor-slot-port-min", 1024); |
| Boolean nimbusDaemon = (Boolean) clusterConf.getOrDefault("nimbus-daemon", false); |
| try { |
| return new LocalCluster.Builder() |
| .withSupervisors(supervisors.intValue()) |
| .withDaemonConf(conf) |
| .withPortsPerSupervisor(ports.intValue()) |
| .withINimbus(inimbus) |
| .withSupervisorSlotPortMin(portMin) |
| .withNimbusDaemon(nimbusDaemon) |
| .build(); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| /** |
| * Run with a local cluster. |
| * |
| * @deprecated use ``` |
| * try (LocalCluster cluster = new LocalCluster.Builder().withSimulatedTime().build()) { |
| * ... |
| * } |
| * ``` |
| * @param code what to run |
| */ |
| @Deprecated |
| public static void withSimulatedTimeLocalCluster(TestJob code) { |
| withSimulatedTimeLocalCluster(new MkClusterParam(), code); |
| } |
| |
| /** |
| * Run with a local cluster. |
| * |
| * @deprecated use ``` |
| * try (LocalCluster cluster = new LocalCluster.Builder().withSimulatedTime()....build()) { |
| * ... |
| * } |
| * ``` |
| * @param param configs to set in the cluster |
| * @param code what to run |
| */ |
| @Deprecated |
| public static void withSimulatedTimeLocalCluster(MkClusterParam param, TestJob code) { |
| try (LocalCluster lc = cluster(param, true)) { |
| code.run(lc); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| /** |
| * Run with a local cluster. |
| * |
| * @deprecated use ``` |
| * try (LocalCluster cluster = new LocalCluster.Builder().withTracked().build()) { |
| * ... |
| * } |
| * ``` |
| * @param code what to run |
| */ |
| @Deprecated |
| public static void withTrackedCluster(TestJob code) { |
| withTrackedCluster(new MkClusterParam(), code); |
| } |
| |
| /** |
| * Run with a local tracked cluster. |
| * |
| * @deprecated use ``` |
| * try (LocalCluster cluster = new LocalCluster.Builder().withTracked()....build()) { |
| * ... |
| * } |
| * ``` |
| * @param param configs to set in the cluster |
| * @param code what to run |
| */ |
| @Deprecated |
| public static void withTrackedCluster(MkClusterParam param, TestJob code) { |
| try (LocalCluster lc = cluster(param, Utils.uuid(), true)) { |
| code.run(lc); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| } |
| |
| /** |
| * In a tracked topology some metrics are tracked. This provides a way to get those metrics. |
| * This is intended mostly for internal testing. |
| * |
| * @param id the id of the tracked cluster |
| * @param key the name of the metric to get. |
| * @return the metric |
| */ |
| @SuppressWarnings("unchecked") |
| @Deprecated |
| public static int globalAmt(String id, String key) { |
| LOG.warn("Reading tracked metrics for ID {}", id); |
| return ((ConcurrentHashMap<String, AtomicInteger>) RegisteredGlobalState.getState(id)).get(key).get(); |
| } |
| |
| /** |
| * Track and capture a topology. |
| * This is intended mostly for internal testing. |
| */ |
| public static CapturedTopology<TrackedTopology> trackAndCaptureTopology(ILocalCluster cluster, StormTopology topology) { |
| CapturedTopology<StormTopology> captured = captureTopology(topology); |
| return new CapturedTopology<>(new TrackedTopology(captured.topology, cluster), captured.capturer); |
| } |
| |
| /** |
| * Rewrites a topology so that all the tuples flowing through it are captured. |
| * @param topology the topology to rewrite |
| * @return the modified topology and a new Bolt that can retrieve the |
| * captured tuples. |
| */ |
| public static CapturedTopology<StormTopology> captureTopology(StormTopology topology) { |
| topology = topology.deepCopy(); //Don't modify the original |
| |
| TupleCaptureBolt capturer = new TupleCaptureBolt(); |
| Map<GlobalStreamId, Grouping> captureBoltInputs = new HashMap<>(); |
| for (Map.Entry<String, SpoutSpec> spoutEntry : topology.get_spouts().entrySet()) { |
| String id = spoutEntry.getKey(); |
| for (Entry<String, StreamInfo> streamEntry : spoutEntry.getValue().get_common().get_streams().entrySet()) { |
| String stream = streamEntry.getKey(); |
| StreamInfo info = streamEntry.getValue(); |
| if (info.is_direct()) { |
| captureBoltInputs.put(new GlobalStreamId(id, stream), Thrift.prepareDirectGrouping()); |
| } else { |
| captureBoltInputs.put(new GlobalStreamId(id, stream), Thrift.prepareGlobalGrouping()); |
| } |
| } |
| } |
| |
| for (Entry<String, Bolt> boltEntry : topology.get_bolts().entrySet()) { |
| String id = boltEntry.getKey(); |
| for (Entry<String, StreamInfo> streamEntry : boltEntry.getValue().get_common().get_streams().entrySet()) { |
| String stream = streamEntry.getKey(); |
| StreamInfo info = streamEntry.getValue(); |
| if (info.is_direct()) { |
| captureBoltInputs.put(new GlobalStreamId(id, stream), Thrift.prepareDirectGrouping()); |
| } else { |
| captureBoltInputs.put(new GlobalStreamId(id, stream), Thrift.prepareGlobalGrouping()); |
| } |
| } |
| } |
| topology.put_to_bolts(Utils.uuid(), new Bolt(Thrift.serializeComponentObject(capturer), |
| Thrift.prepareComponentCommon(captureBoltInputs, new HashMap<>(), null))); |
| return new CapturedTopology<>(topology, capturer); |
| } |
| |
| /** |
| * Run a topology to completion capturing all of the messages that are emitted. This only works when all of the spouts are |
| * instances of {@link org.apache.storm.testing.CompletableSpout}. |
| * @param cluster the cluster to submit the topology to |
| * @param topology the topology itself |
| * @return a map of the component to the list of tuples it emitted |
| * @throws TException on any error from nimbus |
| */ |
| public static Map<String, List<FixedTuple>> completeTopology(ILocalCluster cluster, StormTopology topology) throws InterruptedException, |
| TException { |
| return completeTopology(cluster, topology, new CompleteTopologyParam()); |
| } |
| |
| /** |
| * Run a topology to completion capturing all of the messages that are emitted. This only works when all of the spouts are |
| * instances of {@link org.apache.storm.testing.CompletableSpout} or are overwritten by MockedSources in param |
| * @param cluster the cluster to submit the topology to |
| * @param topology the topology itself |
| * @param param parameters to describe how to complete a topology |
| * @return a map of the component to the list of tuples it emitted |
| * @throws TException on any error from nimbus. |
| */ |
| public static Map<String, List<FixedTuple>> completeTopology(ILocalCluster cluster, StormTopology topology, |
| CompleteTopologyParam param) throws TException, InterruptedException { |
| Map<String, List<FixedTuple>> ret = null; |
| CapturedTopology<StormTopology> capTopo = captureTopology(topology); |
| topology = capTopo.topology; |
| String topoName = param.getTopologyName(); |
| if (topoName == null) { |
| topoName = "topologytest-" + Utils.uuid(); |
| } |
| |
| Map<String, SpoutSpec> spouts = topology.get_spouts(); |
| MockedSources ms = param.getMockedSources(); |
| if (ms != null) { |
| for (Entry<String, List<FixedTuple>> mocked : ms.getData().entrySet()) { |
| FixedTupleSpout newSpout = new FixedTupleSpout(mocked.getValue()); |
| spouts.get(mocked.getKey()).set_spout_object(Thrift.serializeComponentObject(newSpout)); |
| } |
| } |
| List<Object> spoutObjects = spouts.values() |
| .stream() |
| .map((spec) -> Thrift.deserializeComponentObject(spec.get_spout_object())) |
| .collect(Collectors.toList()); |
| |
| for (Object o : spoutObjects) { |
| if (!(o instanceof CompletableSpout)) { |
| throw new RuntimeException( |
| "Cannot complete topology unless every spout is a CompletableSpout (or mocked to be); failed by " + o); |
| } |
| } |
| |
| for (Object spout : spoutObjects) { |
| ((CompletableSpout) spout).startup(); |
| } |
| |
| cluster.submitTopology(topoName, param.getStormConf(), topology); |
| |
| if (Time.isSimulating()) { |
| cluster.advanceClusterTime(11); |
| } |
| |
| IStormClusterState state = cluster.getClusterState(); |
| String topoId = state.getTopoId(topoName).get(); |
| //Give the topology time to come up without using it to wait for the spouts to complete |
| simulateWait(cluster); |
| Integer timeoutMs = param.getTimeoutMs(); |
| if (timeoutMs == null) { |
| timeoutMs = TEST_TIMEOUT_MS; |
| } |
| whileTimeout(timeoutMs, |
| () -> !isEvery(spoutObjects, (o) -> ((CompletableSpout) o).isExhausted()), |
| () -> { |
| try { |
| simulateWait(cluster); |
| } catch (Exception e) { |
| throw new RuntimeException(); |
| } |
| }); |
| |
| KillOptions killOpts = new KillOptions(); |
| killOpts.set_wait_secs(0); |
| cluster.killTopologyWithOpts(topoName, killOpts); |
| |
| whileTimeout(timeoutMs, |
| () -> state.assignmentInfo(topoId, null) != null, |
| () -> { |
| try { |
| simulateWait(cluster); |
| } catch (Exception e) { |
| throw new RuntimeException(); |
| } |
| }); |
| |
| if (param.getCleanupState()) { |
| for (Object o : spoutObjects) { |
| ((CompletableSpout) o).clean(); |
| } |
| ret = capTopo.capturer.getAndRemoveResults(); |
| } else { |
| ret = capTopo.capturer.getAndClearResults(); |
| } |
| |
| return ret; |
| } |
| |
| /** |
| * If using simulated time simulate waiting for 10 seconds. This is intended for internal testing only. |
| */ |
| public static void simulateWait(ILocalCluster cluster) throws InterruptedException { |
| if (Time.isSimulating()) { |
| cluster.advanceClusterTime(10); |
| Thread.sleep(100); |
| } |
| } |
| |
| /** |
| * Get all of the tuples from a given component on the default stream. |
| * @param results the results of running a completed topology |
| * @param componentId the id of the component to look at |
| * @return a list of the tuple values. |
| */ |
| public static List<List<Object>> readTuples(Map<String, List<FixedTuple>> results, String componentId) { |
| return readTuples(results, componentId, Utils.DEFAULT_STREAM_ID); |
| } |
| |
| /** |
| * Get all of the tuples from a given component on a given stream. |
| * @param results the results of running a completed topology |
| * @param componentId the id of the component to look at |
| * @param streamId the id of the stream to look for. |
| * @return a list of the tuple values. |
| */ |
| public static List<List<Object>> readTuples(Map<String, List<FixedTuple>> results, String componentId, String streamId) { |
| List<List<Object>> ret = new ArrayList<>(); |
| List<FixedTuple> streamResult = results.get(componentId); |
| if (streamResult != null) { |
| for (FixedTuple tuple : streamResult) { |
| if (streamId.equals(tuple.stream)) { |
| ret.add(tuple.values); |
| } |
| } |
| } |
| return ret; |
| } |
| |
| /** |
| * Create a tracked topology. |
| * @deprecated use {@link org.apache.storm.testing.TrackedTopology} directly. |
| */ |
| @Deprecated |
| public static TrackedTopology mkTrackedTopology(ILocalCluster cluster, StormTopology topology) { |
| return new TrackedTopology(topology, cluster); |
| } |
| |
| /** |
| * Simulated time wait for a tracked topology. This is intended for internal testing. |
| */ |
| public static void trackedWait(CapturedTopology<TrackedTopology> topo) { |
| topo.topology.trackedWait(); |
| } |
| |
| /** |
| * Simulated time wait for a tracked topology. This is intended for internal testing. |
| */ |
| public static void trackedWait(CapturedTopology<TrackedTopology> topo, Integer amt) { |
| topo.topology.trackedWait(amt); |
| } |
| |
| /** |
| * Simulated time wait for a tracked topology. This is intended for internal testing. |
| */ |
| public static void trackedWait(CapturedTopology<TrackedTopology> topo, Integer amt, Integer timeoutMs) { |
| topo.topology.trackedWait(amt, timeoutMs); |
| } |
| |
| /** |
| * Simulated time wait for a tracked topology. This is intended for internal testing. |
| */ |
| public static void trackedWait(TrackedTopology topo) { |
| topo.trackedWait(); |
| } |
| |
| /** |
| * Simulated time wait for a tracked topology. This is intended for internal testing. |
| */ |
| public static void trackedWait(TrackedTopology topo, Integer amt) { |
| topo.trackedWait(amt); |
| } |
| |
| /** |
| * Simulated time wait for a tracked topology. This is intended for internal testing. |
| */ |
| public static void trackedWait(TrackedTopology topo, Integer amt, Integer timeoutMs) { |
| topo.trackedWait(amt, timeoutMs); |
| } |
| |
| /** |
| * Simulated time wait for a cluster. This is intended for internal testing. |
| */ |
| public static void advanceClusterTime(ILocalCluster cluster, Integer secs) throws InterruptedException { |
| advanceClusterTime(cluster, secs, 1); |
| } |
| |
| /** |
| * Simulated time wait for a cluster. This is intended for internal testing. |
| */ |
| public static void advanceClusterTime(ILocalCluster cluster, Integer secs, Integer step) throws InterruptedException { |
| cluster.advanceClusterTime(secs, step); |
| } |
| |
| /** |
| * Count how many times each element appears in the Collection. |
| * @param c a collection of values |
| * @return a map of the unique values in c to the count of those values. |
| */ |
| public static <T> Map<T, Integer> multiset(Collection<T> c) { |
| Map<T, Integer> ret = new HashMap<T, Integer>(); |
| for (T t : c) { |
| Integer i = ret.get(t); |
| if (i == null) { |
| i = new Integer(0); |
| } |
| i += 1; |
| ret.put(t, i); |
| } |
| return ret; |
| } |
| |
| private static void printRec(Object o, String prefix) { |
| if (o instanceof Collection) { |
| LOG.info("{} {} ({}) [", prefix, o, o.getClass()); |
| for (Object sub : (Collection) o) { |
| printRec(sub, prefix + " "); |
| } |
| LOG.info("{} ]", prefix); |
| } else if (o instanceof Map) { |
| Map<?, ?> m = (Map<?, ?>) o; |
| LOG.info("{} {} ({}) {", prefix, o, o.getClass()); |
| for (Map.Entry<?, ?> entry : m.entrySet()) { |
| printRec(entry.getKey(), prefix + " "); |
| LOG.info("{} ->", prefix); |
| printRec(entry.getValue(), prefix + " "); |
| } |
| LOG.info("{} }", prefix); |
| } else { |
| LOG.info("{} {} ({})", prefix, o, o.getClass()); |
| } |
| } |
| |
| /** |
| * Check if two collections are equivalent ignoring the order of elements. |
| */ |
| public static <T> boolean multiseteq(Collection<T> a, Collection<T> b) { |
| boolean ret = multiset(a).equals(multiset(b)); |
| if (!ret) { |
| printRec(multiset(a), "MS-A:"); |
| printRec(multiset(b), "MS-B:"); |
| } |
| return ret; |
| } |
| |
| /** |
| * Create a {@link org.apache.storm.tuple.Tuple} for use with testing. |
| * @param values the values to appear in the tuple |
| */ |
| public static Tuple testTuple(List<Object> values) { |
| return testTuple(values, new MkTupleParam()); |
| } |
| |
| /** |
| * Create a {@link org.apache.storm.tuple.Tuple} for use with testing. |
| * @param values the values to appear in the tuple |
| * @param param parametrs describing more details about the tuple |
| */ |
| public static Tuple testTuple(List<Object> values, MkTupleParam param) { |
| String stream = param.getStream(); |
| if (stream == null) { |
| stream = Utils.DEFAULT_STREAM_ID; |
| } |
| |
| String component = param.getComponent(); |
| if (component == null) { |
| component = "component"; |
| } |
| |
| int task = 1; |
| |
| List<String> fields = param.getFields(); |
| if (fields == null) { |
| fields = new ArrayList<>(values.size()); |
| for (int i = 1; i <= values.size(); i++) { |
| fields.add("field" + i); |
| } |
| } |
| |
| Map<Integer, String> taskToComp = new HashMap<>(); |
| taskToComp.put(task, component); |
| Map<String, Map<String, Fields>> compToStreamToFields = new HashMap<>(); |
| Map<String, Fields> streamToFields = new HashMap<>(); |
| streamToFields.put(stream, new Fields(fields)); |
| compToStreamToFields.put(component, streamToFields); |
| |
| TopologyContext context = new TopologyContext(null, |
| ConfigUtils.readStormConfig(), |
| taskToComp, |
| null, |
| compToStreamToFields, |
| null, |
| "test-storm-id", |
| null, |
| null, |
| 1, |
| null, |
| null, |
| new HashMap<>(), |
| new HashMap<>(), |
| new HashMap<>(), |
| new HashMap<>(), |
| new AtomicBoolean(false), |
| null); |
| return new TupleImpl(context, values, component, 1, stream); |
| } |
| |
| /** |
| * Simply produces a boolean to see if a specific state is true or false. |
| */ |
| public interface Condition { |
| boolean exec(); |
| } |
| |
| /** |
| * A topology that has all messages captured and can be read later on. |
| * This is intended mostly for internal testing. |
| * @param <T> the topology (tracked or regular) |
| */ |
| public static final class CapturedTopology<T> { |
| public final T topology; |
| /** |
| * a Bolt that will hold all of the captured data. |
| */ |
| public final TupleCaptureBolt capturer; |
| |
| public CapturedTopology(T topology, TupleCaptureBolt capturer) { |
| this.topology = topology; |
| this.capturer = capturer; |
| } |
| } |
| } |
