markpollack

markpollack

public class TopTags implements Processor<String, String> {

    private static final Logger LOG = LoggerFactory.getLogger(TopTags.class);
    // a Java Chronicle backed persistent map that auto delete on JVM shutdown for testing purposes
    // each update on the map will trigger subscriber signals
    MapStream<String, Integer> persistentMap;

    public TopTags() throws IOException {
        this.persistentMap = IOStreams.persistentMap("popularTags", true);
    }

markpollack

package com.acme;

import org.junit.*;

import static org.junit.Assert.*;
import static org.springframework.xd.dirt.test.process.SingleNodeProcessingChainSupport.*;

import org.springframework.xd.dirt.server.SingleNodeApplication;
import org.springframework.xd.dirt.test.SingleNodeIntegrationTestSupport;
import org.springframework.xd.dirt.test.SingletonModuleRegistry;

markpollack

antlr\:antlr=2.7.7
aopalliance\:aopalliance=1.0
bsf\:bsf=2.4.0
ch.qos.logback\:logback-classic=1.1.2
com.beust\:jcommander=1.35
com.caucho\:hessian=4.0.7
com.codahale.metrics\:metrics-core=3.0.2
com.codahale.metrics\:metrics-ganglia=3.0.2
com.codahale.metrics\:metrics-graphite=3.0.2
com.codahale.metrics\:metrics-servlets=3.0.2

markpollack

package spring.xd.rxjava;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.ResolvableType;
import org.springframework.integration.handler.AbstractMessageProducingHandler;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;

markpollack

public class SensorProcessor implements ReactorProcessor<SensorData, SensorSummary> {



    @Override
    public Stream<SensorSummary> process(Stream<SensorData> inputStream) {

        return inputStream
                .window(5, 20, TimeUnit.SECONDS)
                //would be better to convert to stream of double 'values' and then have generic avg for type safety.

markpollack

    /**
     * Create a new Subject, if not already present in thread local storage.
     * @param data
     */
    void process2(String data) {

        System.out.println(Thread.currentThread().getName() + "> Received data in handler " + data);

        
        BehaviorSubject<Object> subject = createSubjectIfNeeded();

markpollack

public interface ReactorModule {

    /**
     * A Stream of messages from the message bus or in the case of composed modules, from the
     * previous XD module.
     * @param inputStream Input stream of Messages
     * @return Output stream of messages sent to the message bus or the next module in the case of
     * composed modules.
     */
    Stream<Message> process(Stream<Message> inputStream);

markpollack

package spring.xd.sample;

import reactor.core.Environment;
import reactor.core.Reactor;
import reactor.core.spec.Reactors;
import reactor.event.Event;
import reactor.event.selector.Selectors;
import reactor.function.Consumer;

markpollack

<?xml version="1.0" encoding="UTF-8"?>
<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
  <modelVersion>4.0.0</modelVersion>
  <groupId>org.springframework.xd</groupId>
  <artifactId>spring-xd-analytics</artifactId>
  <version>1.0.0.BUILD-SNAPSHOT</version>
  <name>Spring XD Anayltics</name>
  <description>Spring XD Anayltics</description>
  <url>https://github.com/spring-projects/spring-xd</url>