Example for Tim Cook of the apache/beam channel

ContainsExampleTest.java

package com.opencore.demo;

import com.google.common.collect.ImmutableList;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.values.PCollection;
import org.junit.Rule;
import org.junit.Test;

import java.io.*;
import java.util.Objects;

/**
 * An example of contains in any order for the slack channel.
 */
public class ContainsExampleTest implements Serializable {

  /**
   * A custom container object.
   */
  public  static class Message implements Serializable {
    private int id;
    private String payload;

    public Message(int id, String payload) {
      this.id = id;
      this.payload = payload;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;
      Message message = (Message) o;
      return id == message.id &&
          Objects.equals(payload, message.payload);
    }

    @Override
    public int hashCode() {
      return Objects.hash(id, payload);
    }
  }

  @Rule
  public final transient TestPipeline p = TestPipeline.create();

  @Test
  public void execute() {

    // source data 
    PCollection<String> input = p.apply(Create.of("hello", "world"));

    // convert to a custom object
    PCollection<Message> messages = input.apply(ParDo.of(new DoFn<String, Message>() {
      @ProcessElement
      public void processElement(ProcessContext c) {
        // create a custom object, and use a predictable ID strategy
        c.output(new Message(c.element().hashCode(), c.element()));
      }
    }));

    // note reversed order (and reuse of ID strategy here)
    ImmutableList<Message> expected = ImmutableList.of(
        new Message("world".hashCode(), "world"),
        new Message("hello".hashCode(), "hello"));

    PAssert.that(messages).containsInAnyOrder(expected);
    p.run();
  }
}