/*
 * Copyright (C) 2008, Robin Rosenberg <robin.rosenberg@dewire.com>
 * Copyright (C) 2008, Shawn O. Pearce <spearce@spearce.org> and others
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Distribution License v. 1.0 which is available at
 * https://www.eclipse.org/org/documents/edl-v10.php.
 *
 * SPDX-License-Identifier: BSD-3-Clause
 */

package org.eclipse.jgit.transport;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.eclipse.jgit.transport.SideBandOutputStream.HDR_SIZE;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Writer;
import java.text.MessageFormat;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.eclipse.jgit.errors.PackProtocolException;
import org.eclipse.jgit.errors.TransportException;
import org.eclipse.jgit.internal.JGitText;
import org.eclipse.jgit.lib.ProgressMonitor;
import org.eclipse.jgit.util.IO;
import org.eclipse.jgit.util.RawParseUtils;

Unmultiplexes the data portion of a side-band channel.

Reading from this input stream obtains data from channel 1, which is typically the bulk data stream.

Channel 2 is transparently unpacked and "scraped" to update a progress monitor. The scraping is performed behind the scenes as part of any of the read methods offered by this stream.

Channel 3 results in an exception being thrown, as the remote side has issued an unrecoverable error.

See Also:
  • SideBandOutputStream
Since:4.11
/** * Unmultiplexes the data portion of a side-band channel. * <p> * Reading from this input stream obtains data from channel 1, which is * typically the bulk data stream. * <p> * Channel 2 is transparently unpacked and "scraped" to update a progress * monitor. The scraping is performed behind the scenes as part of any of the * read methods offered by this stream. * <p> * Channel 3 results in an exception being thrown, as the remote side has issued * an unrecoverable error. * * @see SideBandOutputStream * @since 4.11 */
public class SideBandInputStream extends InputStream { static final int CH_DATA = 1; static final int CH_PROGRESS = 2; static final int CH_ERROR = 3; private static Pattern P_UNBOUNDED = Pattern .compile("^([\\w ]+): +(\\d+)(?:, done\\.)? *[\r\n]$"); //$NON-NLS-1$ private static Pattern P_BOUNDED = Pattern .compile("^([\\w ]+): +\\d+% +\\( *(\\d+)/ *(\\d+)\\)(?:, done\\.)? *[\r\n]$"); //$NON-NLS-1$ private final InputStream rawIn; private final PacketLineIn pckIn; private final ProgressMonitor monitor; private final Writer messages; private final OutputStream out; private String progressBuffer = ""; //$NON-NLS-1$ private String currentTask; private int lastCnt; private boolean eof; private int channel; private int available; SideBandInputStream(final InputStream in, final ProgressMonitor progress, final Writer messageStream, OutputStream outputStream) { rawIn = in; pckIn = new PacketLineIn(rawIn); monitor = progress; messages = messageStream; currentTask = ""; //$NON-NLS-1$ out = outputStream; }
{@inheritDoc}
/** {@inheritDoc} */
@Override public int read() throws IOException { needDataPacket(); if (eof) return -1; available--; return rawIn.read(); }
{@inheritDoc}
/** {@inheritDoc} */
@Override public int read(byte[] b, int off, int len) throws IOException { int r = 0; while (len > 0) { needDataPacket(); if (eof) break; final int n = rawIn.read(b, off, Math.min(len, available)); if (n < 0) break; r += n; off += n; len -= n; available -= n; } return eof && r == 0 ? -1 : r; } private void needDataPacket() throws IOException { if (eof || (channel == CH_DATA && available > 0)) return; for (;;) { available = pckIn.readLength(); if (available == 0) { eof = true; return; } channel = rawIn.read() & 0xff; available -= HDR_SIZE; // length header plus channel indicator if (available == 0) continue; switch (channel) { case CH_DATA: return; case CH_PROGRESS: progress(readString(available)); continue; case CH_ERROR: eof = true; throw new TransportException(remote(readString(available))); default: throw new PackProtocolException( MessageFormat.format(JGitText.get().invalidChannel, Integer.valueOf(channel))); } } } private void progress(String pkt) throws IOException { pkt = progressBuffer + pkt; for (;;) { final int lf = pkt.indexOf('\n'); final int cr = pkt.indexOf('\r'); final int s; if (0 <= lf && 0 <= cr) s = Math.min(lf, cr); else if (0 <= lf) s = lf; else if (0 <= cr) s = cr; else break; doProgressLine(pkt.substring(0, s + 1)); pkt = pkt.substring(s + 1); } progressBuffer = pkt; } private void doProgressLine(String msg) throws IOException { Matcher matcher; matcher = P_BOUNDED.matcher(msg); if (matcher.matches()) { final String taskname = matcher.group(1); if (!currentTask.equals(taskname)) { currentTask = taskname; lastCnt = 0; beginTask(Integer.parseInt(matcher.group(3))); } final int cnt = Integer.parseInt(matcher.group(2)); monitor.update(cnt - lastCnt); lastCnt = cnt; return; } matcher = P_UNBOUNDED.matcher(msg); if (matcher.matches()) { final String taskname = matcher.group(1); if (!currentTask.equals(taskname)) { currentTask = taskname; lastCnt = 0; beginTask(ProgressMonitor.UNKNOWN); } final int cnt = Integer.parseInt(matcher.group(2)); monitor.update(cnt - lastCnt); lastCnt = cnt; return; } messages.write(msg); if (out != null) out.write(msg.getBytes(UTF_8)); } private void beginTask(int totalWorkUnits) { monitor.beginTask(remote(currentTask), totalWorkUnits); } private static String remote(String msg) { String prefix = JGitText.get().prefixRemote; StringBuilder r = new StringBuilder(prefix.length() + msg.length() + 1); r.append(prefix); if (prefix.length() > 0 && prefix.charAt(prefix.length() - 1) != ' ') { r.append(' '); } r.append(msg); return r.toString(); } private String readString(int len) throws IOException { final byte[] raw = new byte[len]; IO.readFully(rawIn, raw, 0, len); return RawParseUtils.decode(UTF_8, raw, 0, len); } }