BasePackFetchConnection.java

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

  13. package org.eclipse.jgit.transport;

  15. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_DELIM;
  16. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_DEEPEN;
  17. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_DEEPEN_NOT;
  18. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_DEEPEN_SINCE;
  19. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_DONE;
  20. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_END;
  21. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_ERR;
  22. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_HAVE;
  23. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_SHALLOW;
  24. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_UNSHALLOW;
  25. import static org.eclipse.jgit.transport.GitProtocolConstants.PACKET_WANT;

  27. import java.io.IOException;
  28. import java.io.InputStream;
  29. import java.io.OutputStream;
  30. import java.text.MessageFormat;
  31. import java.time.Instant;
  32. import java.util.Arrays;
  33. import java.util.Collection;
  34. import java.util.Collections;
  35. import java.util.Date;
  36. import java.util.HashSet;
  37. import java.util.LinkedHashSet;
  38. import java.util.List;
  39. import java.util.Set;

  41. import org.eclipse.jgit.errors.PackProtocolException;
  42. import org.eclipse.jgit.errors.RemoteRepositoryException;
  43. import org.eclipse.jgit.errors.TransportException;
  44. import org.eclipse.jgit.internal.JGitText;
  45. import org.eclipse.jgit.lib.AnyObjectId;
  46. import org.eclipse.jgit.lib.Config;
  47. import org.eclipse.jgit.lib.MutableObjectId;
  48. import org.eclipse.jgit.lib.NullProgressMonitor;
  49. import org.eclipse.jgit.lib.ObjectDatabase;
  50. import org.eclipse.jgit.lib.ObjectId;
  51. import org.eclipse.jgit.lib.ObjectInserter;
  52. import org.eclipse.jgit.lib.ProgressMonitor;
  53. import org.eclipse.jgit.lib.Ref;
  54. import org.eclipse.jgit.revwalk.RevCommit;
  55. import org.eclipse.jgit.revwalk.RevCommitList;
  56. import org.eclipse.jgit.revwalk.RevFlag;
  57. import org.eclipse.jgit.revwalk.RevObject;
  58. import org.eclipse.jgit.revwalk.RevSort;
  59. import org.eclipse.jgit.revwalk.RevWalk;
  60. import org.eclipse.jgit.revwalk.filter.CommitTimeRevFilter;
  61. import org.eclipse.jgit.revwalk.filter.RevFilter;
  62. import org.eclipse.jgit.transport.GitProtocolConstants.MultiAck;
  63. import org.eclipse.jgit.transport.PacketLineIn.AckNackResult;
  64. import org.eclipse.jgit.util.StringUtils;
  65. import org.eclipse.jgit.util.TemporaryBuffer;

  67. /**
  68.  * Fetch implementation using the native Git pack transfer service.
  69.  * <p>
  70.  * This is the canonical implementation for transferring objects from the remote
  71.  * repository to the local repository by talking to the 'git-upload-pack'
  72.  * service. Objects are packed on the remote side into a pack file and then sent
  73.  * down the pipe to us.
  74.  * <p>
  75.  * This connection requires only a bi-directional pipe or socket, and thus is
  76.  * easily wrapped up into a local process pipe, anonymous TCP socket, or a
  77.  * command executed through an SSH tunnel.
  78.  * <p>
  79.  * If {@link org.eclipse.jgit.transport.BasePackConnection#statelessRPC} is
  80.  * {@code true}, this connection can be tunneled over a request-response style
  81.  * RPC system like HTTP. The RPC call boundary is determined by this class
  82.  * switching from writing to the OutputStream to reading from the InputStream.
  83.  * <p>
  84.  * Concrete implementations should just call
  85.  * {@link #init(java.io.InputStream, java.io.OutputStream)} and
  86.  * {@link #readAdvertisedRefs()} methods in constructor or before any use. They
  87.  * should also handle resources releasing in {@link #close()} method if needed.
  88.  */
  89. public abstract class BasePackFetchConnection extends BasePackConnection
  90.         implements FetchConnection {
  91.     /**
  92.      * Maximum number of 'have' lines to send before giving up.
  93.      * <p>
  94.      * During {@link #negotiate(ProgressMonitor, boolean, Set)} we send at most this many
  95.      * commits to the remote peer as 'have' lines without an ACK response before
  96.      * we give up.
  97.      */
  98.     private static final int MAX_HAVES = 256;

  100.     /**
  101.      * Amount of data the client sends before starting to read.
  102.      * <p>
  103.      * Any output stream given to the client must be able to buffer this many
  104.      * bytes before the client will stop writing and start reading from the
  105.      * input stream. If the output stream blocks before this many bytes are in
  106.      * the send queue, the system will deadlock.
  107.      */
  108.     protected static final int MIN_CLIENT_BUFFER = 2 * 32 * 46 + 8;

  110.     /**
  111.      * Include tags if we are also including the referenced objects.
  112.      * @since 2.0
  113.      */
  114.     public static final String OPTION_INCLUDE_TAG = GitProtocolConstants.OPTION_INCLUDE_TAG;

  116.     /**
  117.      * Multi-ACK support for improved negotiation.
  118.      * @since 2.0
  119.      */
  120.     public static final String OPTION_MULTI_ACK = GitProtocolConstants.OPTION_MULTI_ACK;

  122.     /**
  123.      * Multi-ACK detailed support for improved negotiation.
  124.      * @since 2.0
  125.      */
  126.     public static final String OPTION_MULTI_ACK_DETAILED = GitProtocolConstants.OPTION_MULTI_ACK_DETAILED;

  128.     /**
  129.      * The client supports packs with deltas but not their bases.
  130.      * @since 2.0
  131.      */
  132.     public static final String OPTION_THIN_PACK = GitProtocolConstants.OPTION_THIN_PACK;

  134.     /**
  135.      * The client supports using the side-band for progress messages.
  136.      * @since 2.0
  137.      */
  138.     public static final String OPTION_SIDE_BAND = GitProtocolConstants.OPTION_SIDE_BAND;

  140.     /**
  141.      * The client supports using the 64K side-band for progress messages.
  142.      * @since 2.0
  143.      */
  144.     public static final String OPTION_SIDE_BAND_64K = GitProtocolConstants.OPTION_SIDE_BAND_64K;

  146.     /**
  147.      * The client supports packs with OFS deltas.
  148.      * @since 2.0
  149.      */
  150.     public static final String OPTION_OFS_DELTA = GitProtocolConstants.OPTION_OFS_DELTA;

  152.     /**
  153.      * The client supports shallow fetches.
  154.      * @since 2.0
  155.      */
  156.     public static final String OPTION_SHALLOW = GitProtocolConstants.OPTION_SHALLOW;

  158.     /**
  159.      * The client does not want progress messages and will ignore them.
  160.      * @since 2.0
  161.      */
  162.     public static final String OPTION_NO_PROGRESS = GitProtocolConstants.OPTION_NO_PROGRESS;

  164.     /**
  165.      * The client supports receiving a pack before it has sent "done".
  166.      * @since 2.0
  167.      */
  168.     public static final String OPTION_NO_DONE = GitProtocolConstants.OPTION_NO_DONE;

  170.     /**
  171.      * The client supports fetching objects at the tip of any ref, even if not
  172.      * advertised.
  173.      * @since 3.1
  174.      */
  175.     public static final String OPTION_ALLOW_TIP_SHA1_IN_WANT = GitProtocolConstants.OPTION_ALLOW_TIP_SHA1_IN_WANT;

  177.     /**
  178.      * The client supports fetching objects that are reachable from a tip of a
  179.      * ref that is allowed to fetch.
  180.      * @since 4.1
  181.      */
  182.     public static final String OPTION_ALLOW_REACHABLE_SHA1_IN_WANT = GitProtocolConstants.OPTION_ALLOW_REACHABLE_SHA1_IN_WANT;

  184.     /**
  185.      * The client specified a filter expression.
  186.      *
  187.      * @since 5.0
  188.      */
  189.     public static final String OPTION_FILTER = GitProtocolConstants.OPTION_FILTER;

  191.     private final RevWalk walk;

  193.     /** All commits that are immediately reachable by a local ref. */
  194.     private RevCommitList<RevCommit> reachableCommits;

  196.     /** Marks an object as having all its dependencies. */
  197.     final RevFlag REACHABLE;

  199.     /** Marks a commit known to both sides of the connection. */
  200.     final RevFlag COMMON;

  202.     /** Like {@link #COMMON} but means its also in {@link #pckState}. */
  203.     private final RevFlag STATE;

  205.     /** Marks a commit listed in the advertised refs. */
  206.     final RevFlag ADVERTISED;

  208.     private MultiAck multiAck = MultiAck.OFF;

  210.     private boolean thinPack;

  212.     private boolean sideband;

  214.     private boolean includeTags;

  216.     private boolean allowOfsDelta;

  218.     private boolean noDone;

  220.     private boolean noProgress;

  222.     private String lockMessage;

  224.     private PackLock packLock;

  226.     private int maxHaves;

  228.     private Integer depth;

  230.     private Instant deepenSince;

  232.     private List<String> deepenNots;

  234.     /**
  235.      * RPC state, if {@link BasePackConnection#statelessRPC} is true or protocol
  236.      * V2 is used.
  237.      */
  238.     private TemporaryBuffer.Heap state;

  240.     private PacketLineOut pckState;

  242.     /**
  243.      * Either FilterSpec.NO_FILTER for a filter that doesn't filter
  244.      * anything, or a filter that indicates what and what not to send to the
  245.      * server.
  246.      */
  247.     private final FilterSpec filterSpec;

  249.     /**
  250.      * Create a new connection to fetch using the native git transport.
  251.      *
  252.      * @param packTransport
  253.      *            the transport.
  254.      */
  255.     public BasePackFetchConnection(PackTransport packTransport) {
  256.         super(packTransport);

  258.         if (local != null) {
  259.             final FetchConfig cfg = getFetchConfig();
  260.             allowOfsDelta = cfg.allowOfsDelta;
  261.             maxHaves = cfg.maxHaves;
  262.         } else {
  263.             allowOfsDelta = true;
  264.             maxHaves = Integer.MAX_VALUE;
  265.         }

  267.         includeTags = transport.getTagOpt() != TagOpt.NO_TAGS;
  268.         thinPack = transport.isFetchThin();
  269.         filterSpec = transport.getFilterSpec();
  270.         depth = transport.getDepth();
  271.         deepenSince = transport.getDeepenSince();
  272.         deepenNots = transport.getDeepenNots();

  274.         if (local != null) {
  275.             walk = new RevWalk(local);
  276.             walk.setRetainBody(false);
  277.             reachableCommits = new RevCommitList<>();
  278.             REACHABLE = walk.newFlag("REACHABLE"); //$NON-NLS-1$
  279.             COMMON = walk.newFlag("COMMON"); //$NON-NLS-1$
  280.             STATE = walk.newFlag("STATE"); //$NON-NLS-1$
  281.             ADVERTISED = walk.newFlag("ADVERTISED"); //$NON-NLS-1$

  283.             walk.carry(COMMON);
  284.             walk.carry(REACHABLE);
  285.             walk.carry(ADVERTISED);
  286.         } else {
  287.             walk = null;
  288.             REACHABLE = null;
  289.             COMMON = null;
  290.             STATE = null;
  291.             ADVERTISED = null;
  292.         }
  293.     }

  295.     static class FetchConfig {
  296.         final boolean allowOfsDelta;

  298.         final int maxHaves;

  300.         FetchConfig(Config c) {
  301.             allowOfsDelta = c.getBoolean("repack", "usedeltabaseoffset", true); //$NON-NLS-1$ //$NON-NLS-2$
  302.             maxHaves = c.getInt("fetch", "maxhaves", Integer.MAX_VALUE); //$NON-NLS-1$ //$NON-NLS-2$
  303.         }

  305.         FetchConfig(boolean allowOfsDelta, int maxHaves) {
  306.             this.allowOfsDelta = allowOfsDelta;
  307.             this.maxHaves = maxHaves;
  308.         }
  309.     }

  311.     /** {@inheritDoc} */
  312.     @Override
  313.     public final void fetch(final ProgressMonitor monitor,
  314.             final Collection<Ref> want, final Set<ObjectId> have)
  315.             throws TransportException {
  316.         fetch(monitor, want, have, null);
  317.     }

  319.     /** {@inheritDoc} */
  320.     @Override
  321.     public final void fetch(final ProgressMonitor monitor,
  322.             final Collection<Ref> want, final Set<ObjectId> have,
  323.             OutputStream outputStream) throws TransportException {
  324.         markStartedOperation();
  325.         doFetch(monitor, want, have, outputStream);
  326.     }

  328.     /** {@inheritDoc} */
  329.     @Override
  330.     public boolean didFetchIncludeTags() {
  331.         return false;
  332.     }

  334.     /** {@inheritDoc} */
  335.     @Override
  336.     public boolean didFetchTestConnectivity() {
  337.         return false;
  338.     }

  340.     /** {@inheritDoc} */
  341.     @Override
  342.     public void setPackLockMessage(String message) {
  343.         lockMessage = message;
  344.     }

  346.     /** {@inheritDoc} */
  347.     @Override
  348.     public Collection<PackLock> getPackLocks() {
  349.         if (packLock != null)
  350.             return Collections.singleton(packLock);
  351.         return Collections.<PackLock> emptyList();
  352.     }

  354.     private void clearState() {
  355.         walk.dispose();
  356.         reachableCommits = null;
  357.         state = null;
  358.         pckState = null;
  359.     }

  361.     /**
  362.      * Execute common ancestor negotiation and fetch the objects.
  363.      *
  364.      * @param monitor
  365.      *            progress monitor to receive status updates. If the monitor is
  366.      *            the {@link org.eclipse.jgit.lib.NullProgressMonitor#INSTANCE}, then the no-progress
  367.      *            option enabled.
  368.      * @param want
  369.      *            the advertised remote references the caller wants to fetch.
  370.      * @param have
  371.      *            additional objects to assume that already exist locally. This
  372.      *            will be added to the set of objects reachable from the
  373.      *            destination repository's references.
  374.      * @param outputStream
  375.      *            ouputStream to write sideband messages to
  376.      * @throws org.eclipse.jgit.errors.TransportException
  377.      *             if any exception occurs.
  378.      * @since 3.0
  379.      */
  380.     protected void doFetch(final ProgressMonitor monitor,
  381.             final Collection<Ref> want, final Set<ObjectId> have,
  382.             OutputStream outputStream) throws TransportException {
  383.         try {
  384.             noProgress = monitor == NullProgressMonitor.INSTANCE;

  386.             markRefsAdvertised();
  387.             markReachable(have, maxTimeWanted(want));

  389.             if (TransferConfig.ProtocolVersion.V2
  390.                     .equals(getProtocolVersion())) {
  391.                 // Protocol V2 always is a "stateless" protocol, even over a
  392.                 // bidirectional pipe: the server serves one "fetch" request and
  393.                 // then forgets anything it has learned, so the next fetch
  394.                 // request has to re-send all wants and previously determined
  395.                 // common objects as "have"s again.
  396.                 state = new TemporaryBuffer.Heap(Integer.MAX_VALUE);
  397.                 pckState = new PacketLineOut(state);
  398.                 try {
  399.                     doFetchV2(monitor, want, outputStream);
  400.                 } finally {
  401.                     clearState();
  402.                 }
  403.                 return;
  404.             }
  405.             // Protocol V0/1
  406.             if (statelessRPC) {
  407.                 state = new TemporaryBuffer.Heap(Integer.MAX_VALUE);
  408.                 pckState = new PacketLineOut(state);
  409.             }
  410.             PacketLineOut output = statelessRPC ? pckState : pckOut;
  411.             if (sendWants(want, output)) {
  412.                 boolean mayHaveShallow = depth != null || deepenSince != null || !deepenNots.isEmpty();
  413.                 Set<ObjectId> shallowCommits = local.getObjectDatabase().getShallowCommits();
  414.                 if (isCapableOf(GitProtocolConstants.CAPABILITY_SHALLOW)) {
  415.                     sendShallow(shallowCommits, output);
  416.                 } else if (mayHaveShallow) {
  417.                     throw new PackProtocolException(JGitText.get().shallowNotSupported);
  418.                 }
  419.                 output.end();
  420.                 outNeedsEnd = false;

  422.                 negotiate(monitor, mayHaveShallow, shallowCommits);

  424.                 clearState();

  426.                 receivePack(monitor, outputStream);
  427.             }
  428.         } catch (CancelledException ce) {
  429.             close();
  430.             return; // Caller should test (or just know) this themselves.
  431.         } catch (IOException | RuntimeException err) {
  432.             close();
  433.             throw new TransportException(err.getMessage(), err);
  434.         }
  435.     }

  437.     private void doFetchV2(ProgressMonitor monitor, Collection<Ref> want,
  438.             OutputStream outputStream) throws IOException, CancelledException {
  439.         sideband = true;
  440.         negotiateBegin();

  442.         pckState.writeString("command=" + GitProtocolConstants.COMMAND_FETCH); //$NON-NLS-1$
  443.         // Capabilities are sent as command arguments in protocol V2
  444.         String agent = UserAgent.get();
  445.         if (agent != null && isCapableOf(GitProtocolConstants.OPTION_AGENT)) {
  446.             pckState.writeString(
  447.                     GitProtocolConstants.OPTION_AGENT + '=' + agent);
  448.         }
  449.         Set<String> capabilities = new HashSet<>();
  450.         String advertised = getCapability(GitProtocolConstants.COMMAND_FETCH);
  451.         if (!StringUtils.isEmptyOrNull(advertised)) {
  452.             capabilities.addAll(Arrays.asList(advertised.split("\\s+"))); //$NON-NLS-1$
  453.         }
  454.         // Arguments
  455.         pckState.writeDelim();
  456.         for (String capability : getCapabilitiesV2(capabilities)) {
  457.             pckState.writeString(capability);
  458.         }

  460.         if (!sendWants(want, pckState)) {
  461.             // We already have everything we wanted.
  462.             return;
  463.         }

  465.         Set<ObjectId> shallowCommits = local.getObjectDatabase().getShallowCommits();
  466.         if (capabilities.contains(GitProtocolConstants.CAPABILITY_SHALLOW)) {
  467.             sendShallow(shallowCommits, pckState);
  468.         } else if (depth != null || deepenSince != null || !deepenNots.isEmpty()) {
  469.             throw new PackProtocolException(JGitText.get().shallowNotSupported);
  470.         }
  471.         // If we send something, we always close it properly ourselves.
  472.         outNeedsEnd = false;

  474.         FetchStateV2 fetchState = new FetchStateV2();
  475.         boolean sentDone = false;
  476.         for (;;) {
  477.             // The "state" buffer contains the full fetch request with all
  478.             // common objects found so far.
  479.             state.writeTo(out, monitor);
  480.             sentDone = sendNextHaveBatch(fetchState, pckOut, monitor);
  481.             if (sentDone) {
  482.                 break;
  483.             }
  484.             if (readAcknowledgments(fetchState, pckIn, monitor)) {
  485.                 // We got a "ready": next should be a patch file.
  486.                 break;
  487.             }
  488.             // Note: C git reads and requires here (and after a packfile) a
  489.             // "0002" packet in stateless RPC transports (https). This "response
  490.             // end" packet is even mentioned in the protocol V2 technical
  491.             // documentation. However, it is not actually part of the public
  492.             // protocol; it occurs only in an internal protocol wrapper in the C
  493.             // git implementation.
  494.         }
  495.         clearState();
  496.         String line = pckIn.readString();
  497.         // If we sent a done, we may have an error reply here.
  498.         if (sentDone && line.startsWith(PACKET_ERR)) {
  499.             throw new RemoteRepositoryException(uri, line.substring(4));
  500.         }

  502.         if (GitProtocolConstants.SECTION_SHALLOW_INFO.equals(line)) {
  503.             line = handleShallowUnshallow(shallowCommits, pckIn);
  504.             if (!PacketLineIn.isDelimiter(line)) {
  505.                 throw new PackProtocolException(MessageFormat
  506.                         .format(JGitText.get().expectedGot, PACKET_DELIM,
  507.                                 line));
  508.             }
  509.             line = pckIn.readString();
  510.         }

  512.         // "wanted-refs" and "packfile-uris" would have to be
  513.         // handled here in that order.
  514.         if (!GitProtocolConstants.SECTION_PACKFILE.equals(line)) {
  515.             throw new PackProtocolException(
  516.                     MessageFormat.format(JGitText.get().expectedGot,
  517.                             GitProtocolConstants.SECTION_PACKFILE, line));
  518.         }
  519.         receivePack(monitor, outputStream);
  520.     }

  522.     /**
  523.      * Sends the next batch of "have"s and terminates the {@code output}.
  524.      *
  525.      * @param fetchState
  526.      *            is updated with information about the number of items written,
  527.      *            and whether to expect a packfile next
  528.      * @param output
  529.      *            to write to
  530.      * @param monitor
  531.      *            for progress reporting and cancellation
  532.      * @return {@code true} if a "done" was written and we should thus expect a
  533.      *         packfile next
  534.      * @throws IOException
  535.      *             on errors
  536.      * @throws CancelledException
  537.      *             on cancellation
  538.      */
  539.     private boolean sendNextHaveBatch(FetchStateV2 fetchState,
  540.             PacketLineOut output, ProgressMonitor monitor)
  541.             throws IOException, CancelledException {
  542.         long n = 0;
  543.         while (n < fetchState.havesToSend) {
  544.             final RevCommit c = walk.next();
  545.             if (c == null) {
  546.                 break;
  547.             }
  548.             output.writeString(PACKET_HAVE + c.getId().name() + '\n');
  549.             n++;
  550.             if (n % 10 == 0 && monitor.isCancelled()) {
  551.                 throw new CancelledException();
  552.             }
  553.         }
  554.         fetchState.havesTotal += n;
  555.         if (n == 0
  556.                 || (fetchState.hadAcks
  557.                         && fetchState.havesWithoutAck > MAX_HAVES)
  558.                 || fetchState.havesTotal > maxHaves) {
  559.             output.writeString(PACKET_DONE + '\n');
  560.             output.end();
  561.             return true;
  562.         }
  563.         // Increment only after the test above. Of course we have no ACKs yet
  564.         // for the newly added "have"s, so it makes no sense to count them
  565.         // against the MAX_HAVES limit.
  566.         fetchState.havesWithoutAck += n;
  567.         output.end();
  568.         fetchState.incHavesToSend(statelessRPC);
  569.         return false;
  570.     }

  572.     /**
  573.      * Reads and processes acknowledgments, adding ACKed objects as "have"s to
  574.      * the global state {@link TemporaryBuffer}.
  575.      *
  576.      * @param fetchState
  577.      *            to update
  578.      * @param input
  579.      *            to read from
  580.      * @param monitor
  581.      *            for progress reporting and cancellation
  582.      * @return {@code true} if a "ready" was received and a packfile is expected
  583.      *         next
  584.      * @throws IOException
  585.      *             on errors
  586.      * @throws CancelledException
  587.      *             on cancellation
  588.      */
  589.     private boolean readAcknowledgments(FetchStateV2 fetchState,
  590.             PacketLineIn input, ProgressMonitor monitor)
  591.             throws IOException, CancelledException {
  592.         String line = input.readString();
  593.         if (!GitProtocolConstants.SECTION_ACKNOWLEDGMENTS.equals(line)) {
  594.             throw new PackProtocolException(MessageFormat.format(
  595.                     JGitText.get().expectedGot,
  596.                     GitProtocolConstants.SECTION_ACKNOWLEDGMENTS, line));
  597.         }
  598.         MutableObjectId returnedId = new MutableObjectId();
  599.         line = input.readString();
  600.         boolean gotReady = false;
  601.         long n = 0;
  602.         while (!PacketLineIn.isEnd(line) && !PacketLineIn.isDelimiter(line)) {
  603.             AckNackResult ack = PacketLineIn.parseACKv2(line, returnedId);
  604.             // If we got a "ready", we just skip the remaining lines after
  605.             // having checked them for being valid. (Normally, the "ready"
  606.             // should be the last line anyway.)
  607.             if (!gotReady) {
  608.                 if (ack == AckNackResult.ACK_COMMON) {
  609.                     // markCommon appends the object to the "state"
  610.                     markCommon(walk.parseAny(returnedId), ack, true);
  611.                     fetchState.havesWithoutAck = 0;
  612.                     fetchState.hadAcks = true;
  613.                 } else if (ack == AckNackResult.ACK_READY) {
  614.                     gotReady = true;
  615.                 }
  616.             }
  617.             n++;
  618.             if (n % 10 == 0 && monitor.isCancelled()) {
  619.                 throw new CancelledException();
  620.             }
  621.             line = input.readString();
  622.         }
  623.         if (gotReady) {
  624.             if (!PacketLineIn.isDelimiter(line)) {
  625.                 throw new PackProtocolException(MessageFormat
  626.                         .format(JGitText.get().expectedGot, PACKET_DELIM,
  627.                                 line));
  628.             }
  629.         } else if (!PacketLineIn.isEnd(line)) {
  630.             throw new PackProtocolException(MessageFormat
  631.                     .format(JGitText.get().expectedGot, PACKET_END, line));
  632.         }
  633.         return gotReady;
  634.     }

  636.     /** {@inheritDoc} */
  637.     @Override
  638.     public void close() {
  639.         if (walk != null)
  640.             walk.close();
  641.         super.close();
  642.     }

  644.     FetchConfig getFetchConfig() {
  645.         return local.getConfig().get(FetchConfig::new);
  646.     }

  648.     private int maxTimeWanted(Collection<Ref> wants) {
  649.         int maxTime = 0;
  650.         for (Ref r : wants) {
  651.             try {
  652.                 final RevObject obj = walk.parseAny(r.getObjectId());
  653.                 if (obj instanceof RevCommit) {
  654.                     final int cTime = ((RevCommit) obj).getCommitTime();
  655.                     if (maxTime < cTime)
  656.                         maxTime = cTime;
  657.                 }
  658.             } catch (IOException error) {
  659.                 // We don't have it, but we want to fetch (thus fixing error).
  660.             }
  661.         }
  662.         return maxTime;
  663.     }

  665.     private void markReachable(Set<ObjectId> have, int maxTime)
  666.             throws IOException {
  667.         for (Ref r : local.getRefDatabase().getRefs()) {
  668.             ObjectId id = r.getPeeledObjectId();
  669.             if (id == null)
  670.                 id = r.getObjectId();
  671.             if (id == null)
  672.                 continue;
  673.             parseReachable(id);
  674.         }

  676.         for (ObjectId id : local.getAdditionalHaves())
  677.             parseReachable(id);

  679.         for (ObjectId id : have)
  680.             parseReachable(id);

  682.         if (maxTime > 0) {
  683.             // Mark reachable commits until we reach maxTime. These may
  684.             // wind up later matching up against things we want and we
  685.             // can avoid asking for something we already happen to have.
  686.             //
  687.             final Date maxWhen = new Date(maxTime * 1000L);
  688.             walk.sort(RevSort.COMMIT_TIME_DESC);
  689.             walk.markStart(reachableCommits);
  690.             walk.setRevFilter(CommitTimeRevFilter.after(maxWhen));
  691.             for (;;) {
  692.                 final RevCommit c = walk.next();
  693.                 if (c == null)
  694.                     break;
  695.                 if (c.has(ADVERTISED) && !c.has(COMMON)) {
  696.                     // This is actually going to be a common commit, but
  697.                     // our peer doesn't know that fact yet.
  698.                     //
  699.                     c.add(COMMON);
  700.                     c.carry(COMMON);
  701.                     reachableCommits.add(c);
  702.                 }
  703.             }
  704.         }
  705.     }

  707.     private void parseReachable(ObjectId id) {
  708.         try {
  709.             RevCommit o = walk.parseCommit(id);
  710.             if (!o.has(REACHABLE)) {
  711.                 o.add(REACHABLE);
  712.                 reachableCommits.add(o);
  713.             }
  714.         } catch (IOException readError) {
  715.             // If we cannot read the value of the ref skip it.
  716.         }
  717.     }

  719.     private boolean sendWants(Collection<Ref> want, PacketLineOut p)
  720.             throws IOException {
  721.         boolean first = true;
  722.         for (Ref r : want) {
  723.             ObjectId objectId = r.getObjectId();
  724.             if (objectId == null) {
  725.                 continue;
  726.             }
  727.             // if depth is set we need to fetch the objects even if they are already available
  728.             if (transport.getDepth() == null) {
  729.                 try {
  730.                     if (walk.parseAny(objectId).has(REACHABLE)) {
  731.                         // We already have this object. Asking for it is
  732.                         // not a very good idea.
  733.                         //
  734.                         continue;
  735.                     }
  736.                 } catch (IOException err) {
  737.                     // Its OK, we don't have it, but we want to fix that
  738.                     // by fetching the object from the other side.
  739.                 }
  740.             }

  742.             final StringBuilder line = new StringBuilder(46);
  743.             line.append(PACKET_WANT).append(objectId.name());
  744.             if (first && TransferConfig.ProtocolVersion.V0
  745.                     .equals(getProtocolVersion())) {
  746.                 line.append(enableCapabilities());
  747.             }
  748.             first = false;
  749.             line.append('\n');
  750.             p.writeString(line.toString());
  751.         }
  752.         if (first) {
  753.             return false;
  754.         }
  755.         if (!filterSpec.isNoOp()) {
  756.             p.writeString(filterSpec.filterLine());
  757.         }
  758.         return true;
  759.     }

  761.     private Set<String> getCapabilitiesV2(Set<String> advertisedCapabilities)
  762.             throws TransportException {
  763.         Set<String> capabilities = new LinkedHashSet<>();
  764.         // Protocol V2 is implicitly capable of all these.
  765.         if (noProgress) {
  766.             capabilities.add(OPTION_NO_PROGRESS);
  767.         }
  768.         if (includeTags) {
  769.             capabilities.add(OPTION_INCLUDE_TAG);
  770.         }
  771.         if (allowOfsDelta) {
  772.             capabilities.add(OPTION_OFS_DELTA);
  773.         }
  774.         if (thinPack) {
  775.             capabilities.add(OPTION_THIN_PACK);
  776.         }
  777.         if (!filterSpec.isNoOp()
  778.                 && !advertisedCapabilities.contains(OPTION_FILTER)) {
  779.             throw new PackProtocolException(uri,
  780.                     JGitText.get().filterRequiresCapability);
  781.         }
  782.         // The FilterSpec will be added later in sendWants().
  783.         return capabilities;
  784.     }

  786.     private String enableCapabilities() throws TransportException {
  787.         final StringBuilder line = new StringBuilder();
  788.         if (noProgress)
  789.             wantCapability(line, OPTION_NO_PROGRESS);
  790.         if (includeTags)
  791.             includeTags = wantCapability(line, OPTION_INCLUDE_TAG);
  792.         if (allowOfsDelta)
  793.             wantCapability(line, OPTION_OFS_DELTA);

  795.         if (wantCapability(line, OPTION_MULTI_ACK_DETAILED)) {
  796.             multiAck = MultiAck.DETAILED;
  797.             if (statelessRPC)
  798.                 noDone = wantCapability(line, OPTION_NO_DONE);
  799.         } else if (wantCapability(line, OPTION_MULTI_ACK))
  800.             multiAck = MultiAck.CONTINUE;
  801.         else
  802.             multiAck = MultiAck.OFF;

  804.         if (thinPack)
  805.             thinPack = wantCapability(line, OPTION_THIN_PACK);
  806.         if (wantCapability(line, OPTION_SIDE_BAND_64K))
  807.             sideband = true;
  808.         else if (wantCapability(line, OPTION_SIDE_BAND))
  809.             sideband = true;

  811.         if (statelessRPC && multiAck != MultiAck.DETAILED) {
  812.             // Our stateless RPC implementation relies upon the detailed
  813.             // ACK status to tell us common objects for reuse in future
  814.             // requests.  If its not enabled, we can't talk to the peer.
  815.             //
  816.             throw new PackProtocolException(uri, MessageFormat.format(
  817.                     JGitText.get().statelessRPCRequiresOptionToBeEnabled,
  818.                     OPTION_MULTI_ACK_DETAILED));
  819.         }

  821.         if (!filterSpec.isNoOp() && !wantCapability(line, OPTION_FILTER)) {
  822.             throw new PackProtocolException(uri,
  823.                     JGitText.get().filterRequiresCapability);
  824.         }

  826.         addUserAgentCapability(line);
  827.         return line.toString();
  828.     }

  830.     private void negotiate(ProgressMonitor monitor, boolean mayHaveShallow, Set<ObjectId> shallowCommits)
  831.             throws IOException, CancelledException {
  832.         final MutableObjectId ackId = new MutableObjectId();
  833.         int resultsPending = 0;
  834.         int havesSent = 0;
  835.         int havesSinceLastContinue = 0;
  836.         boolean receivedContinue = false;
  837.         boolean receivedAck = false;
  838.         boolean receivedReady = false;

  840.         if (statelessRPC) {
  841.             state.writeTo(out, null);
  842.         }

  844.         negotiateBegin();
  845.         SEND_HAVES: for (;;) {
  846.             final RevCommit c = walk.next();
  847.             if (c == null) {
  848.                 break SEND_HAVES;
  849.             }

  851.             ObjectId o = c.getId();
  852.             pckOut.writeString(PACKET_HAVE + o.name() + '\n');
  853.             havesSent++;
  854.             havesSinceLastContinue++;

  856.             if ((31 & havesSent) != 0) {
  857.                 // We group the have lines into blocks of 32, each marked
  858.                 // with a flush (aka end). This one is within a block so
  859.                 // continue with another have line.
  860.                 //
  861.                 continue;
  862.             }

  864.             if (monitor.isCancelled()) {
  865.                 throw new CancelledException();
  866.             }

  868.             pckOut.end();
  869.             resultsPending++; // Each end will cause a result to come back.

  871.             if (havesSent == 32 && !statelessRPC) {
  872.                 // On the first block we race ahead and try to send
  873.                 // more of the second block while waiting for the
  874.                 // remote to respond to our first block request.
  875.                 // This keeps us one block ahead of the peer.
  876.                 //
  877.                 continue;
  878.             }

  880.             READ_RESULT: for (;;) {
  881.                 final AckNackResult anr = pckIn.readACK(ackId);
  882.                 switch (anr) {
  883.                 case NAK:
  884.                     // More have lines are necessary to compute the
  885.                     // pack on the remote side. Keep doing that.
  886.                     //
  887.                     resultsPending--;
  888.                     break READ_RESULT;

  890.                 case ACK:
  891.                     // The remote side is happy and knows exactly what
  892.                     // to send us. There is no further negotiation and
  893.                     // we can break out immediately.
  894.                     //
  895.                     multiAck = MultiAck.OFF;
  896.                     resultsPending = 0;
  897.                     receivedAck = true;
  898.                     if (statelessRPC) {
  899.                         state.writeTo(out, null);
  900.                     }
  901.                     break SEND_HAVES;

  903.                 case ACK_CONTINUE:
  904.                 case ACK_COMMON:
  905.                 case ACK_READY:
  906.                     // The server knows this commit (ackId). We don't
  907.                     // need to send any further along its ancestry, but
  908.                     // we need to continue to talk about other parts of
  909.                     // our local history.
  910.                     //
  911.                     markCommon(walk.parseAny(ackId), anr, statelessRPC);
  912.                     receivedAck = true;
  913.                     receivedContinue = true;
  914.                     havesSinceLastContinue = 0;
  915.                     if (anr == AckNackResult.ACK_READY) {
  916.                         receivedReady = true;
  917.                     }
  918.                     break;
  919.                 }

  921.                 if (monitor.isCancelled()) {
  922.                     throw new CancelledException();
  923.                 }
  924.             }

  926.             if (noDone && receivedReady) {
  927.                 break SEND_HAVES;
  928.             }
  929.             if (statelessRPC) {
  930.                 state.writeTo(out, null);
  931.             }

  933.             if ((receivedContinue && havesSinceLastContinue > MAX_HAVES)
  934.                     || havesSent >= maxHaves) {
  935.                 // Our history must be really different from the remote's.
  936.                 // We just sent a whole slew of have lines, and it did not
  937.                 // recognize any of them. Avoid sending our entire history
  938.                 // to them by giving up early.
  939.                 //
  940.                 break SEND_HAVES;
  941.             }
  942.         }

  944.         // Tell the remote side we have run out of things to talk about.
  945.         //
  946.         if (monitor.isCancelled()) {
  947.             throw new CancelledException();
  948.         }

  950.         if (!receivedReady || !noDone) {
  951.             // When statelessRPC is true we should always leave SEND_HAVES
  952.             // loop above while in the middle of a request. This allows us
  953.             // to just write done immediately.
  954.             //
  955.             pckOut.writeString(PACKET_DONE + '\n');
  956.             pckOut.flush();
  957.         }

  959.         if (!receivedAck) {
  960.             // Apparently if we have never received an ACK earlier
  961.             // there is one more result expected from the done we
  962.             // just sent to the remote.
  963.             //
  964.             multiAck = MultiAck.OFF;
  965.             resultsPending++;
  966.         }

  968.         if (mayHaveShallow) {
  969.             String line = handleShallowUnshallow(shallowCommits, pckIn);
  970.             if (!PacketLineIn.isEnd(line)) {
  971.                 throw new PackProtocolException(MessageFormat
  972.                         .format(JGitText.get().expectedGot, PACKET_END, line));
  973.             }
  974.         }

  976.         READ_RESULT: while (resultsPending > 0 || multiAck != MultiAck.OFF) {
  977.             final AckNackResult anr = pckIn.readACK(ackId);
  978.             resultsPending--;
  979.             switch (anr) {
  980.             case NAK:
  981.                 // A NAK is a response to an end we queued earlier
  982.                 // we eat it and look for another ACK/NAK message.
  983.                 //
  984.                 break;

  986.             case ACK:
  987.                 // A solitary ACK at this point means the remote won't
  988.                 // speak anymore, but is going to send us a pack now.
  989.                 //
  990.                 break READ_RESULT;

  992.             case ACK_CONTINUE:
  993.             case ACK_COMMON:
  994.             case ACK_READY:
  995.                 // We will expect a normal ACK to break out of the loop.
  996.                 //
  997.                 multiAck = MultiAck.CONTINUE;
  998.                 break;
  999.             }

  1001.             if (monitor.isCancelled()) {
  1002.                 throw new CancelledException();
  1003.             }
  1004.         }
  1005.     }

  1007.     private void negotiateBegin() throws IOException {
  1008.         walk.resetRetain(REACHABLE, ADVERTISED);
  1009.         walk.markStart(reachableCommits);
  1010.         walk.sort(RevSort.COMMIT_TIME_DESC);
  1011.         walk.setRevFilter(new RevFilter() {
  1012.             @Override
  1013.             public RevFilter clone() {
  1014.                 return this;
  1015.             }

  1017.             @Override
  1018.             public boolean include(RevWalk walker, RevCommit c) {
  1019.                 final boolean remoteKnowsIsCommon = c.has(COMMON);
  1020.                 if (c.has(ADVERTISED)) {
  1021.                     // Remote advertised this, and we have it, hence common.
  1022.                     // Whether or not the remote knows that fact is tested
  1023.                     // before we added the flag. If the remote doesn't know
  1024.                     // we have to still send them this object.
  1025.                     //
  1026.                     c.add(COMMON);
  1027.                 }
  1028.                 return !remoteKnowsIsCommon;
  1029.             }

  1031.             @Override
  1032.             public boolean requiresCommitBody() {
  1033.                 return false;
  1034.             }
  1035.         });
  1036.     }

  1038.     private void markRefsAdvertised() {
  1039.         for (Ref r : getRefs()) {
  1040.             markAdvertised(r.getObjectId());
  1041.             if (r.getPeeledObjectId() != null)
  1042.                 markAdvertised(r.getPeeledObjectId());
  1043.         }
  1044.     }

  1046.     private void markAdvertised(AnyObjectId id) {
  1047.         try {
  1048.             walk.parseAny(id).add(ADVERTISED);
  1049.         } catch (IOException readError) {
  1050.             // We probably just do not have this object locally.
  1051.         }
  1052.     }

  1054.     private void markCommon(RevObject obj, AckNackResult anr, boolean useState)
  1055.             throws IOException {
  1056.         if (useState && anr == AckNackResult.ACK_COMMON && !obj.has(STATE)) {
  1057.             pckState.writeString(PACKET_HAVE + obj.name() + '\n');
  1058.             obj.add(STATE);
  1059.         }
  1060.         obj.add(COMMON);
  1061.         if (obj instanceof RevCommit)
  1062.             ((RevCommit) obj).carry(COMMON);
  1063.     }

  1065.     private void receivePack(final ProgressMonitor monitor,
  1066.             OutputStream outputStream) throws IOException {
  1067.         onReceivePack();
  1068.         InputStream input = in;
  1069.         SideBandInputStream sidebandIn = null;
  1070.         if (sideband) {
  1071.             sidebandIn = new SideBandInputStream(input, monitor,
  1072.                     getMessageWriter(), outputStream);
  1073.             input = sidebandIn;
  1074.         }

  1076.         try (ObjectInserter ins = local.newObjectInserter()) {
  1077.             PackParser parser = ins.newPackParser(input);
  1078.             parser.setAllowThin(thinPack);
  1079.             parser.setObjectChecker(transport.getObjectChecker());
  1080.             parser.setLockMessage(lockMessage);
  1081.             packLock = parser.parse(monitor);
  1082.             ins.flush();
  1083.         } finally {
  1084.             if (sidebandIn != null) {
  1085.                 sidebandIn.drainMessages();
  1086.             }
  1087.         }
  1088.     }

  1090.     private void sendShallow(Set<ObjectId> shallowCommits, PacketLineOut output)
  1091.             throws IOException {
  1092.         for (ObjectId shallowCommit : shallowCommits) {
  1093.             output.writeString(PACKET_SHALLOW + shallowCommit.name());
  1094.         }

  1096.         if (depth != null) {
  1097.             output.writeString(PACKET_DEEPEN + depth);
  1098.         }

  1100.         if (deepenSince != null) {
  1101.             output.writeString(
  1102.                     PACKET_DEEPEN_SINCE + deepenSince.getEpochSecond());
  1103.         }

  1105.         if (deepenNots != null) {
  1106.             for (String deepenNotRef : deepenNots) {
  1107.                 output.writeString(PACKET_DEEPEN_NOT + deepenNotRef);
  1108.             }
  1109.         }
  1110.     }

  1112.     private String handleShallowUnshallow(
  1113.             Set<ObjectId> advertisedShallowCommits, PacketLineIn input)
  1114.             throws IOException {
  1115.         String line = input.readString();
  1116.         ObjectDatabase objectDatabase = local.getObjectDatabase();
  1117.         HashSet<ObjectId> newShallowCommits = new HashSet<>(
  1118.                 advertisedShallowCommits);
  1119.         while (!PacketLineIn.isDelimiter(line) && !PacketLineIn.isEnd(line)) {
  1120.             if (line.startsWith(PACKET_SHALLOW)) {
  1121.                 newShallowCommits.add(ObjectId
  1122.                         .fromString(line.substring(PACKET_SHALLOW.length())));
  1123.             } else if (line.startsWith(PACKET_UNSHALLOW)) {
  1124.                 ObjectId unshallow = ObjectId
  1125.                         .fromString(line.substring(PACKET_UNSHALLOW.length()));
  1126.                 if (!advertisedShallowCommits.contains(unshallow)) {
  1127.                     throw new PackProtocolException(MessageFormat.format(
  1128.                             JGitText.get().notShallowedUnshallow,
  1129.                             unshallow.name()));
  1130.                 }
  1131.                 newShallowCommits.remove(unshallow);
  1132.             }
  1133.             line = input.readString();
  1134.         }
  1135.         objectDatabase.setShallowCommits(newShallowCommits);
  1136.         return line;
  1137.     }

  1139.     /**
  1140.      * Notification event delivered just before the pack is received from the
  1141.      * network. This event can be used by RPC such as {@link org.eclipse.jgit.transport.TransportHttp} to
  1142.      * disable its request magic and ensure the pack stream is read correctly.
  1143.      *
  1144.      * @since 2.0
  1145.      */
  1146.     protected void onReceivePack() {
  1147.         // By default do nothing for TCP based protocols.
  1148.     }

  1150.     private static class CancelledException extends Exception {
  1151.         private static final long serialVersionUID = 1L;
  1152.     }

  1154.     private static class FetchStateV2 {

  1156.         long havesToSend = 32;

  1158.         long havesTotal;

  1160.         // Set to true if we got at least one ACK in protocol V2.
  1161.         boolean hadAcks;

  1163.         // Counts haves without ACK. Use as cutoff for negotiation only once
  1164.         // hadAcks == true.
  1165.         long havesWithoutAck;

  1167.         void incHavesToSend(boolean statelessRPC) {
  1168.             if (statelessRPC) {
  1169.                 // Increase this quicker since connection setup costs accumulate
  1170.                 if (havesToSend < 16384) {
  1171.                     havesToSend *= 2;
  1172.                 } else {
  1173.                     havesToSend = havesToSend * 11 / 10;
  1174.                 }
  1175.             } else {
  1176.                 havesToSend += 32;
  1177.             }
  1178.         }
  1179.     }
  1180. }
Created with JaCoCo 0.8.7.202105040129