DomainTypePipelines.java

public class DomainTypePipelines {
  
  	@Test
	public void domainTypePipelines() throws Exception {
		RingBufferWorkProcessor<Object> p = RingBufferWorkProcessor.create();
		Stream<Object> s = Streams.wrap(p);

		s.filter(o -> o instanceof Person)
		 .map(Person.class::cast)
		 .consume(person -> System.out.println(Thread.currentThread() + " person: " + person));

		s.filter(o -> o instanceof Dog)
		 .map(Dog.class::cast)
		 .consume(dog -> System.out.println(Thread.currentThread() + " dog: " + dog));

		s.filter(o -> o instanceof Cat)
		 .map(Cat.class::cast)
		 .consume(cat -> System.out.println(Thread.currentThread() + " cat: " + cat));

		p.onNext(new Person("John", "Doe"));
		p.onNext(new Dog("Saint Bernard", "Rufus"));
		p.onNext(new Cat("Calico", "Skippy"));

		Thread.sleep(500);
	}

	private static final class Person {
		String firstname;
		String lastname;

		public Person() {
		}

		public Person(String firstname, String lastname) {
			this.firstname = firstname;
			this.lastname = lastname;
		}

		@Override
		public String toString() {
			return "Person{" +
			       "firstname='" + firstname + '\'' +
			       ", lastname='" + lastname + '\'' +
			       '}';
		}
	}

	private static final class Dog {
		String breed;
		String name;

		public Dog() {
		}

		public Dog(String breed, String name) {
			this.breed = breed;
			this.name = name;
		}

		@Override
		public String toString() {
			return "Dog{" +
			       "breed='" + breed + '\'' +
			       ", name='" + name + '\'' +
			       '}';
		}
	}

	private static final class Cat {
		String breed;
		String name;

		public Cat() {
		}

		public Cat(String breed, String name) {
			this.breed = breed;
			this.name = name;
		}

		@Override
		public String toString() {
			return "Cat{" +
			       "breed='" + breed + '\'' +
			       ", name='" + name + '\'' +
			       '}';
		}
	}

}