jbrisbin

package reactor.groovy.ext

import groovy.transform.CompileStatic
import groovy.util.slurpersupport.GPathResult
import org.reactivestreams.Publisher
import reactor.io.buffer.Buffer
import reactor.rx.Stream
import reactor.rx.Streams

@CompileStatic

jbrisbin

import com.basho.riak.client.core.RiakCluster
import com.basho.riak.client.core.operations.StoreOperation
import com.basho.riak.client.core.query.Location
import com.basho.riak.client.core.query.Namespace
import com.basho.riak.client.core.query.RiakObject
import com.basho.riak.client.core.util.BinaryValue
import org.slf4j.LoggerFactory
import org.springframework.boot.SpringApplication
import org.springframework.boot.autoconfigure.EnableAutoConfiguration
import org.springframework.cloud.stream.annotation.EnableBinding

jbrisbin

@SpringBootApplication
@EnableBinding(Source.class)
class TrackerNetKafkaSource {

    static {
        Environment.initializeIfEmpty().assignErrorJournal()
    }

    static def TFL_URL = "http://cloud.tfl.gov.uk/TrackerNet/PredictionSummary/C"

jbrisbin

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));

jbrisbin

+-( spring-xd (master) ):> ./gradlew clean --stacktrace
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=512m; support was removed in 8.0

FAILURE: Build failed with an exception.

* Where:
Build file '/Users/jbrisbin/Development/Projects/spring-projects/master/spring-xd/build.gradle' line: 850

* What went wrong:
A problem occurred evaluating root project 'spring-xd'.

jbrisbin

public class DeltaAwareExpiringSession implements ExpiringSession {

	private final Queue<Map.Entry<String, Object>> deltas = new LinkedList<Map.Entry<String, Object>>();

	@Id
	private String id                  = UUID.randomUUID().toString();
	private long   creationTime        = System.currentTimeMillis();
	private long   lastAccessedTime    = creationTime;
	private int    maxInactiveInterval = DEFAULT_MAX_INACTIVE_INTERVAL_SECONDS;

jbrisbin

interface Buffer<B> extends Publisher<B> {
	
	<NEWB> Buffer<NEWB> intercept(Interceptor<B, NEWB> fn);

	Buffer<B> append(Buffer<B> buff);

	Buffer<B> append(B obj);

}

jbrisbin

ReactorTcpServer.listen(3000, ByteBuf.class)
                .log("connection")
                .consume(conn -> conn.out(conn.in().log("in"))
                                     .log("confirmation")
                                     .consume(buf -> LOG.info("write confirmed: {}", buf)));

jbrisbin

public interface TcpConnection<R, W> {

  Reader<R> reader();

  Writer<W> writer();

  public interface Reader<R> extends Publisher<R> {
    <NEWR> Reader<NEWR> intercept(Interceptor<R, NEWR> interceptor);
  }

jbrisbin

interface Channel<IN, OUT> extends Publisher<IN> {
	
	Publisher<Boolean> write(Publisher<? extends OUT> data);

}

interface ClientSocketOptions {
	
	SocketAddress connectAddress();