replicate/mdsal-replicate-netty/src/main/java/org/opendaylight/mdsal/replicate/netty/SplittingOutputStream.java

   1 /*
   2  * Copyright (c) 2020 PANTHEON.tech, s.r.o. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.mdsal.replicate.netty;
   9
  10 import static com.google.common.base.Verify.verify;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import io.netty.buffer.ByteBuf;
  14 import io.netty.buffer.Unpooled;
  15 import java.io.IOException;
  16 import java.io.OutputStream;
  17 import java.util.List;
  18 import java.util.Objects;
  19
  20 /**
  21  * An OutputStream which makes sure to slice messages to a maximum size. This prevents array reallocations and
  22  * GC thrashing on huge objects.
  23  */
  24 final class SplittingOutputStream extends OutputStream {
  25     private static final int INIT_BUF = 4096;
  26
  27     static {
  28         verify(INIT_BUF <= Constants.LENGTH_FIELD_MAX);
  29     }
  30
  31     private final List<Object> out;
  32
  33     private ByteBuf buf;
  34
  35     SplittingOutputStream(final List<Object> out) {
  36         this.out = requireNonNull(out);
  37         allocBuffer();
  38     }
  39
  40     @Override
  41     @SuppressWarnings("checkstyle:parameterName")
  42     public void write(final int b) throws IOException {
  43         buf.writeByte(b);
  44         checkSend();
  45     }
  46
  47     @Override
  48     @SuppressWarnings("checkstyle:parameterName")
  49     public void write(final byte[] b, final int off, final int len) throws IOException {
  50         Objects.checkFromIndexSize(off, len, b.length);
  51
  52         int left = len;
  53         int ptr = off;
  54         while (left > 0) {
  55             final int chunk = Math.min(Constants.LENGTH_FIELD_MAX - buf.writerIndex(), left);
  56
  57             buf.writeBytes(b, ptr, chunk);
  58             ptr += chunk;
  59             left -= chunk;
  60             checkSend();
  61         }
  62     }
  63
  64     @Override
  65     public void close() {
  66         if (buf.writerIndex() != 0) {
  67             out.add(buf);
  68         }
  69         buf = null;
  70     }
  71
  72     private void allocBuffer() {
  73         // FIXME: use buffer allocator?
  74         buf = Unpooled.buffer(INIT_BUF, Constants.LENGTH_FIELD_MAX);
  75         buf.writeByte(Constants.MSG_DTC_CHUNK);
  76     }
  77
  78     private void checkSend() {
  79         if (buf.writerIndex() == Constants.LENGTH_FIELD_MAX) {
  80             out.add(buf);
  81             allocBuffer();
  82         }
  83     }
  84 }