CompositeBuffer.java

/*
 * Copyright 2015 Netflix, Inc.
 *
 * 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 io.reactivex.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.Unpooled;
import rx.Observable;
import rx.exceptions.Exceptions;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class CompositeBuffer {

    public static void main(String[] args) throws IOException {

        ByteBuf buf1 = Unpooled.buffer().writeBytes("Hello1".getBytes());
        ByteBuf buf2 = Unpooled.buffer().writeBytes("Hello2".getBytes());
        ByteBuf buf3 = Unpooled.buffer().writeBytes("Hello3".getBytes());

        Observable.just(buf1, buf2, buf3)
                  .collect(Unpooled::compositeBuffer,
                           (composite, buf) -> {
                               int readable = buf.readableBytes();
                               composite.addComponents(buf);
                               composite.writerIndex(composite.writerIndex() + readable);
                           })
                  .map(ByteBufInputStream::new)
                  .map(InputStreamReader::new)
                  .map(BufferedReader::new)
                  .toBlocking()
                  .forEach(reader -> {
                      try {
                          System.out.println(reader.readLine());
                      } catch (IOException e) {
                          throw Exceptions.propagate(e);
                      }
                  });
    }
}