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
  19 /**
  20  * An OutputStream which makes sure to slice messages to a maximum size. This prevents array reallocations and
  21  * GC thrashing on huge objects.
  22  */
  23 final class SplittingOutputStream extends OutputStream {
  24     private static final int INIT_BUF = 4096;
  25
  26     static {
  27         verify(INIT_BUF <= Constants.LENGTH_FIELD_MAX);
  28     }
  29
  30     private final List<Object> out;
  31
  32     private ByteBuf buf;
  33
  34     SplittingOutputStream(final List<Object> out) {
  35         this.out = requireNonNull(out);
  36         allocBuffer();
  37     }
  38
  39     @Override
  40     @SuppressWarnings("checkstyle:parameterName")
  41     public void write(final int b) throws IOException {
  42         buf.writeByte(b);
  43         checkSend();
  44     }
  45
  46     @Override
  47     @SuppressWarnings("checkstyle:parameterName")
  48     public void write(final byte[] b, final int off, final int len) throws IOException {
  49         if (off < 0 || len < 0) {
  50             throw new IndexOutOfBoundsException();
  51         }
  52
  53         int left = len;
  54         int ptr = off;
  55         while (left > 0) {
  56             final int chunk = Math.min(Constants.LENGTH_FIELD_MAX - buf.writerIndex(), left);
  57
  58             buf.writeBytes(b, ptr, chunk);
  59             ptr += chunk;
  60             left -= chunk;
  61             checkSend();
  62         }
  63     }
  64
  65     @Override
  66     public void close() {
  67         if (buf.writerIndex() != 0) {
  68             out.add(buf);
  69         }
  70         buf = null;
  71     }
  72
  73     private void allocBuffer() {
  74         // FIXME: use buffer allocator?
  75         buf = Unpooled.buffer(INIT_BUF, Constants.LENGTH_FIELD_MAX);
  76         buf.writeByte(Constants.MSG_DTC_CHUNK);
  77     }
  78
  79     private void checkSend() {
  80         if (buf.writerIndex() == Constants.LENGTH_FIELD_MAX) {
  81             out.add(buf);
  82             allocBuffer();
  83         }
  84     }
  85 }