001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.juneau.terminal.region;
018
019import static org.apache.juneau.commons.utils.Shorts.*;
020import static org.apache.juneau.terminal.report.TerminalViewChecks.*;
021
022import java.util.*;
023import java.util.concurrent.atomic.*;
024import java.util.logging.*;
025
026/**
027 * A {@link TerminalSource} over a fixed-capacity ring buffer, for producers that run in the same process.
028 *
029 * <p>
030 * The producer calls {@link #write(byte[], int, int)} and finally {@link #finish()}.  Once more than the capacity has
031 * been written, the oldest bytes are overwritten; a reader asking for them gets a {@code truncated} chunk that starts
032 * at the oldest offset still held.  All methods are thread-safe.
033 *
034 * <h5 class='section'>Example:</h5>
035 * <p class='bjava'>
036 *    MemoryTerminalSource <jv>term</jv> = MemoryTerminalSource.<jsm>create</jsm>(4 * 1024 * 1024, 120, 40);
037 *    <jv>term</jv>.write(<jv>bytes</jv>, 0, <jv>bytes</jv>.length);
038 *    <jv>term</jv>.finish();
039 * </p>
040 */
041public final class MemoryTerminalSource implements TerminalSource {
042
043   private static final Logger LOG = Logger.getLogger(MemoryTerminalSource.class.getName());
044
045   private record Nudge(String id, TerminalNudgePublisher publisher) {}
046
047   private final byte[] ring;
048   private final int cols;
049   private final int rows;
050   private long end;
051   private boolean finished;
052   private final AtomicReference<Nudge> nudge = new AtomicReference<>();
053
054   private MemoryTerminalSource(int capacity, int cols, int rows) {
055      ring = new byte[capacity];
056      this.cols = cols;
057      this.rows = rows;
058   }
059
060   /**
061    * Creates a source.
062    *
063    * @param capacity The ring size in bytes, at least 1.
064    * @param cols The producer's terminal width, 1-9999.
065    * @param rows The producer's terminal height, 1-9999.
066    * @return A new source.
067    * @throws IllegalArgumentException If a value is out of range.
068    */
069   public static MemoryTerminalSource create(int capacity, int cols, int rows) {
070      if (capacity < 1)
071         throw iaex("MemoryTerminalSource capacity must be at least 1; got '%s'.", capacity);
072      if (cols < 1 || cols > 9999 || rows < 1 || rows > 9999)
073         throw iaex("MemoryTerminalSource size must be 1-9999 x 1-9999; got '%sx%s'.", cols, rows);
074      return new MemoryTerminalSource(capacity, cols, rows);
075   }
076
077   /**
078    * Reports every append and the finish to a publisher, outside this source's lock.
079    *
080    * <p>
081    * A {@link RuntimeException} thrown by the publisher is logged at {@code FINE} and otherwise ignored, so a write
082    * never fails because of a nudge.
083    *
084    * @param terminalId The terminal id passed to the publisher; must be a valid run id.
085    * @param publisher The publisher.
086    * @return This object.
087    * @throws NullPointerException If an argument is <jk>null</jk>.
088    * @throws IllegalArgumentException If the id is not a valid run id.
089    * @throws IllegalStateException If a publisher is already set.
090    */
091   public MemoryTerminalSource nudges(String terminalId, TerminalNudgePublisher publisher) {
092      Objects.requireNonNull(terminalId, "terminalId");
093      Objects.requireNonNull(publisher, "publisher");
094      if (! isRunId(terminalId))
095         throw iaex("MemoryTerminalSource nudge id is not a valid run id: '%s'.", terminalId);
096      synchronized (this) {
097         if (nudge.get() != null)
098            throw new IllegalStateException("MemoryTerminalSource nudges are already set.");
099         nudge.set(new Nudge(terminalId, publisher));
100      }
101      return this;
102   }
103
104   private void publish(Nudge n, long e, boolean done) {
105      try {
106         n.publisher().nudge(n.id(), e, done);
107      } catch (RuntimeException x) {
108         LOG.log(Level.FINE, x, () -> "Terminal nudge publisher failed for '" + n.id() + "'.");
109      }
110   }
111
112   /**
113    * Appends bytes.
114    *
115    * @param b The bytes.
116    * @param off The start in {@code b}.
117    * @param len The number of bytes.
118    * @throws IllegalStateException If {@link #finish()} has been called.
119    * @throws IndexOutOfBoundsException If {@code off} and {@code len} do not fit in {@code b}.
120    */
121   public void write(byte[] b, int off, int len) {
122      Objects.checkFromIndexSize(off, len, b.length);
123      Nudge n;
124      long e;
125      synchronized (this) {
126         if (finished)
127            throw new IllegalStateException("MemoryTerminalSource is finished.");
128         if (len > ring.length) {
129            off += len - ring.length;
130            end += len - ring.length;
131            len = ring.length;
132         }
133         var p = (int)(end % ring.length);
134         var first = Math.min(len, ring.length - p);
135         System.arraycopy(b, off, ring, p, first);
136         System.arraycopy(b, off + first, ring, 0, len - first);
137         end += len;
138         e = end;
139         n = nudge.get();
140      }
141      if (n != null)
142         publish(n, e, false);
143   }
144
145   /** Marks the stream finished: readers that reach the end see {@code done}. */
146   public void finish() {
147      Nudge n;
148      long e;
149      synchronized (this) {
150         finished = true;
151         e = end;
152         n = nudge.get();
153      }
154      if (n != null)
155         publish(n, e, true);
156   }
157
158   @Override /* TerminalSource */
159   public synchronized TerminalChunk read(long fromOffset, int maxBytes) {
160      if (fromOffset < 0)
161         throw iaex("fromOffset must be at least 0; got '%s'.", fromOffset);
162      if (fromOffset > end)
163         return new TerminalChunk(new byte[0], end, end, finished, cols, rows, false, false);
164      var oldest = Math.max(0, end - ring.length);
165      var truncated = fromOffset < oldest;
166      var start = truncated ? oldest : fromOffset;
167      var len = (int)Math.min(Math.max(0, Math.min(maxBytes, MAX_READ_BYTES)), end - start);
168      var out = new byte[len];
169      var p = (int)(start % ring.length);
170      var first = Math.min(len, ring.length - p);
171      System.arraycopy(ring, p, out, 0, first);
172      System.arraycopy(ring, 0, out, first, len - first);
173      var next = start + len;
174      return new TerminalChunk(out, next, end, finished && next == end, cols, rows, truncated, false);
175   }
176}