JetTest.java

/*
 * Copyright (c) 2008-2016, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package jet;

import com.hazelcast.jet.DAG;
import com.hazelcast.jet.Edge;
import com.hazelcast.jet.Inbox;
import com.hazelcast.jet.Jet;
import com.hazelcast.jet.JetInstance;
import com.hazelcast.jet.Outbox;
import com.hazelcast.jet.Processor;
import com.hazelcast.jet.ProcessorSupplier;
import com.hazelcast.jet.SimpleProcessorSupplier;
import com.hazelcast.jet.Vertex;
import com.hazelcast.jet.impl.connector.IListWriter;
import com.hazelcast.jet.impl.connector.IMapReader;
import com.hazelcast.jet.stream.IStreamMap;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class JetTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        JetInstance jetInstance = Jet.newJetInstance();
        IStreamMap<Object, Object> map = jetInstance.getMap("map");
        for (int i = 0; i < 20; i++) {
            map.put(i, i);
        }

        DAG dag = new DAG();
        Vertex mapReader = new Vertex("mapReader", IMapReader.supplier("map"));
        Vertex myProcessor = new Vertex("myProcessor", new MyProcessorSupplier());
        Vertex listWriter = new Vertex("listWriter", IListWriter.supplier("list"));
        dag.vertex(mapReader);
        dag.vertex(myProcessor);
        dag.vertex(listWriter);

        dag.edge(Edge.between(mapReader, myProcessor).allToOne());
        dag.edge(Edge.between(myProcessor, listWriter));

        Future<Void> future = jetInstance.newJob(dag).execute();
        future.get();

        int size = jetInstance.getList("list").size();
        System.err.println("fatal list size: " + size);
        System.exit(0);
    }

    static class MyProcessorSupplier implements ProcessorSupplier {

        public Processor get() {
            return new Processor() {

                Outbox outbox;

                @Override
                public void init(Outbox outbox) {
                    this.outbox = outbox;
                }

                @Override
                public void process(int ordinal, Inbox inbox) {
                    System.err.println("fatal ordinal: " + ordinal + " this: " + this);
                    int sum = 0;
                    for (Object o; (o = inbox.peek()) != null; ) {
                        Map.Entry entry = (Map.Entry) o;
                        System.err.println("fatal o: " + o);
                        sum += (Integer) (entry.getValue());
                        inbox.remove();
                    }
                    System.err.println("qwe " + sum);
                    outbox.add(sum);
                }
            };
        }

        @Override
        public void init(Context context) {
            System.err.println("asd localParallelism " + context.localParallelism());
        }

        @Override
        public List<Processor> get(int count) {
            List<Processor> processors = new ArrayList<Processor>(count);
            for (int i = 0; i < count; i++) {
                processors.add(get());
            }
            return processors;
        }

        @Override
        public void complete(Throwable error) {
            new Throwable("naber").printStackTrace();
        }
    }
}