jeryini · April 14, 2015 08:46 · jeryini · Apr 14, 2015
diff --git a/SimpleTcpServer.java b/SimpleTcpServer.java
 package com.jernejerin.reactor.samples;

 import reactor.Environment;
 import reactor.fn.Consumer;
 import reactor.io.codec.json.JsonCodec;
 import reactor.io.net.ChannelStream;
 import reactor.io.net.NetStreams;
 import reactor.io.net.tcp.TcpClient;
 import reactor.io.net.tcp.TcpServer;
 import reactor.rx.Streams;

 /**
 * Created by Jernej Jerin on 9.4.2015.
 */
 public class SimpleTcpServer {
    public static void main(String[] args) throws InterruptedException{
        // environment initialization
        Environment.initialize();

        JsonCodec<A, A> codec = new JsonCodec<A, A>(A.class);

        // TCP server
        TcpServer<A, A> server = NetStreams.tcpServer(
            spec -> spec
                .listen(5000)
                .codec(codec)
                .dispatcher(Environment.cachedDispatcher())
        );

        // consumer for TCP server
        server.consume(new Consumer<ChannelStream<A, A>>() {
            @Override
            public void accept(ChannelStream<A, A> channel) {
                channel.consume(new Consumer<A>() {
                    @Override
                    public void accept(A data) {
                        System.out.printf("Receiving data from client -> %s on thread %s%n", data.getId(), Thread.currentThread());
                    }
                });
            }
        });

        server.start().await();

        // TCP client
        TcpClient<A, A> client = NetStreams.tcpClient(
            spec -> spec
                .connect("localhost", 5000)
                .codec(codec)
                .dispatcher(Environment.cachedDispatcher())
        );

        // consumer for Client
        for (int i = 0; i < 10; i++) {
            A a = new A(i);
            client.consume(ch -> ch.sink(Streams.just(a)));
            client.open().await();
        }

        client.close().await();
        server.shutdown().await();

        Thread.sleep(10000000);
    }
 }

 class A {
    private int id;

    A() {}

    A(int id) {
        this.id = id;
    }

    public void setId(int id) {
        this.id = id;
    }

    public int getId() {
        return this.id;
    }
 }
	package com.jernejerin.reactor.samples;

	import reactor.Environment;
	import reactor.fn.Consumer;
	import reactor.io.codec.json.JsonCodec;
	import reactor.io.net.ChannelStream;
	import reactor.io.net.NetStreams;
	import reactor.io.net.tcp.TcpClient;
	import reactor.io.net.tcp.TcpServer;
	import reactor.rx.Streams;

	/**
	* Created by Jernej Jerin on 9.4.2015.
	*/
	public class SimpleTcpServer {
	public static void main(String[] args) throws InterruptedException{
	// environment initialization
	Environment.initialize();

	JsonCodec<A, A> codec = new JsonCodec<A, A>(A.class);

	// TCP server
	TcpServer<A, A> server = NetStreams.tcpServer(
	spec -> spec
	.listen(5000)
	.codec(codec)
	.dispatcher(Environment.cachedDispatcher())
	);

	// consumer for TCP server
	server.consume(new Consumer<ChannelStream<A, A>>() {
	@Override
	public void accept(ChannelStream<A, A> channel) {
	channel.consume(new Consumer<A>() {
	@Override
	public void accept(A data) {
	System.out.printf("Receiving data from client -> %s on thread %s%n", data.getId(), Thread.currentThread());
	}
	});
	}
	});

	server.start().await();

	// TCP client
	TcpClient<A, A> client = NetStreams.tcpClient(
	spec -> spec
	.connect("localhost", 5000)
	.codec(codec)
	.dispatcher(Environment.cachedDispatcher())
	);

	// consumer for Client
	for (int i = 0; i < 10; i++) {
	A a = new A(i);
	client.consume(ch -> ch.sink(Streams.just(a)));
	client.open().await();
	}

	client.close().await();
	server.shutdown().await();

	Thread.sleep(10000000);
	}
	}

	class A {
	private int id;

	A() {}

	A(int id) {
	this.id = id;
	}

	public void setId(int id) {
	this.id = id;
	}

	public int getId() {
	return this.id;
	}
	}
No results found