text: package io.ray.streaming.runtime.demo;
  
  import io.ray.api.Ray;
  import io.ray.streaming.api.context.StreamingContext;
  import io.ray.streaming.api.function.impl.SinkFunction;
  import io.ray.streaming.api.stream.DataStreamSource;
  import java.io.IOException;
  import java.nio.file.Files;
  import java.nio.file.Paths;
  import java.nio.file.StandardOpenOption;
  import java.util.Arrays;
  import java.util.concurrent.TimeUnit;
  import org.apache.commons.lang3.StringUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.testng.Assert;
  import org.testng.annotations.Test;
  
  public class UnionStreamTest {
  
    private static final Logger LOG = LoggerFactory.getLogger(UnionStreamTest.class);
  
    @Test(timeOut = 60000)
    public void testUnionStream() throws Exception {
      Ray.shutdown();
      String sinkFileName = "/tmp/testUnionStream.txt";
      Files.deleteIfExists(Paths.get(sinkFileName));
  
      StreamingContext context = StreamingContext.buildContext();
      DataStreamSource<Integer> streamSource1 =
          DataStreamSource.fromCollection(context, Arrays.asList(1, 1));
      DataStreamSource<Integer> streamSource2 =
          DataStreamSource.fromCollection(context, Arrays.asList(1, 1));
      DataStreamSource<Integer> streamSource3 =
          DataStreamSource.fromCollection(context, Arrays.asList(1, 1));
      streamSource1
          .union(streamSource2, streamSource3)
          .sink((SinkFunction<Integer>) value -> {
            LOG.info("UnionStreamTest, sink: {}", value);
            try {
              if (!Files.exists(Paths.get(sinkFileName))) {
                Files.createFile(Paths.get(sinkFileName));
              }
              Files.write(Paths.get(sinkFileName), value.toString().getBytes(),
                  StandardOpenOption.APPEND);
            } catch (IOException e) {
              throw new RuntimeException(e);
            }
          });
      context.execute("UnionStreamTest");
      int sleptTime = 0;
      TimeUnit.SECONDS.sleep(3);
      while (true) {
        if (Files.exists(Paths.get(sinkFileName))) {
          TimeUnit.SECONDS.sleep(3);
          String text = String.join(", ", Files.readAllLines(Paths.get(sinkFileName)));
          Assert.assertEquals(text, StringUtils.repeat("1", 6));
          LOG.info("Execution succeed");
          break;
        }
        sleptTime += 1;
        if (sleptTime >= 60) {
          throw new RuntimeException("Execution not finished");
        }
        LOG.info("Wait finish...");
        TimeUnit.SECONDS.sleep(1);
      }
      context.stop();
      LOG.info("HybridStreamTest succeed");
    }
  
  }
  