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         if (off < 0 || len < 0) {
  51             throw new IndexOutOfBoundsException();
  52         }
  53
  54         int left = len;
  55         int ptr = off;
  56         while (left > 0) {
  57             final int chunk = Math.min(Constants.LENGTH_FIELD_MAX - buf.writerIndex(), left);
  58
  59             buf.writeBytes(b, ptr, chunk);
  60             ptr += chunk;
  61             left -= chunk;
  62             checkSend();
  63         }
  64     }
  65
  66     @Override
  67     public void close() {
  68         if (buf.writerIndex() != 0) {
  69             out.add(buf);
  70         }
  71         buf = null;
  72     }
  73
  74     private void allocBuffer() {
  75         // FIXME: use buffer allocator?
  76         buf = Unpooled.buffer(INIT_BUF, Constants.LENGTH_FIELD_MAX);
  77         buf.writeByte(Constants.MSG_DTC_CHUNK);
  78     }
  79
  80     private void checkSend() {
  81         if (buf.writerIndex() == Constants.LENGTH_FIELD_MAX) {
  82             out.add(buf);
  83             allocBuffer();
  84         }
  85     }
  86 }