src/BodyPipe.h

   1 #ifndef SQUID_BODY_PIPE_H
   2 #define SQUID_BODY_PIPE_H
   3
   4 #include "MemBuf.h"
   5 #include "base/AsyncJob.h"
   6 #include "base/CbcPointer.h"
   7
   8 class BodyPipe;
   9
  10 /** Interface for those who want to produce body content for others.
  11  * BodyProducer is expected to create the BodyPipe.
  12  * One pipe cannot have more than one producer.
  13  */
  14 class BodyProducer: virtual public AsyncJob
  15 {
  16 public:
  17     typedef CbcPointer<BodyProducer> Pointer;
  18
  19     BodyProducer():AsyncJob("BodyProducer") {}
  20     virtual ~BodyProducer() {}
  21
  22     virtual void noteMoreBodySpaceAvailable(RefCount<BodyPipe> bp) = 0;
  23     virtual void noteBodyConsumerAborted(RefCount<BodyPipe> bp) = 0;
  24
  25 protected:
  26     void stopProducingFor(RefCount<BodyPipe> &pipe, bool atEof);
  27 };
  28
  29 /** Interface for those who want to consume body content from others.
  30  * BodyConsumer is expected to register with an existing BodyPipe
  31  * by calling BodyPipe::setConsumer().
  32  * One pipe cannot have more than one consumer.
  33  */
  34 class BodyConsumer: virtual public AsyncJob
  35 {
  36 public:
  37     typedef CbcPointer<BodyConsumer> Pointer;
  38
  39     BodyConsumer():AsyncJob("BodyConsumer") {}
  40     virtual ~BodyConsumer() {}
  41
  42     virtual void noteMoreBodyDataAvailable(RefCount<BodyPipe> bp) = 0;
  43     virtual void noteBodyProductionEnded(RefCount<BodyPipe> bp) = 0;
  44     virtual void noteBodyProducerAborted(RefCount<BodyPipe> bp) = 0;
  45
  46 protected:
  47     void stopConsumingFrom(RefCount<BodyPipe> &pipe);
  48 };
  49
  50 /** Makes raw buffer checkin/checkout interface efficient and exception-safe.
  51  * Either append or consume operations can be performed on a checkedout buffer.
  52  */
  53 class BodyPipeCheckout
  54 {
  55 public:
  56     friend class BodyPipe;
  57
  58 public:
  59     BodyPipeCheckout(BodyPipe &pipe); // checks out
  60     ~BodyPipeCheckout(); // aborts checkout unless checkedIn
  61
  62     void checkIn();
  63
  64 public:
  65     BodyPipe &pipe;
  66     MemBuf &buf;
  67     const uint64_t offset; // of current content, relative to the body start
  68
  69 protected:
  70     const size_t checkedOutSize;
  71     bool checkedIn;
  72
  73 private:
  74     BodyPipeCheckout(const BodyPipeCheckout &); // prevent copying
  75     BodyPipeCheckout &operator =(const BodyPipeCheckout &); // prevent assignment
  76 };
  77
  78 /** Connects those who produces message body content with those who
  79  * consume it. For example, connects ConnStateData with FtpStateData OR
  80  * ICAPModXact with HttpStateData.
  81  */
  82 class BodyPipe: public RefCountable
  83 {
  84 public:
  85     typedef RefCount<BodyPipe> Pointer;
  86     typedef BodyProducer Producer;
  87     typedef BodyConsumer Consumer;
  88     typedef BodyPipeCheckout Checkout;
  89
  90     enum { MaxCapacity = SQUID_TCP_SO_RCVBUF };
  91
  92     friend class BodyPipeCheckout;
  93
  94 public:
  95     BodyPipe(Producer *aProducer);
  96     ~BodyPipe(); // asserts that producer and consumer are cleared
  97
  98     void setBodySize(uint64_t aSize); // set body size
  99     bool bodySizeKnown() const { return theBodySize >= 0; }
 100     uint64_t bodySize() const;
 101     uint64_t consumedSize() const { return theGetSize; }
 102     bool productionEnded() const { return !theProducer; }
 103
 104     // called by producers
 105     void clearProducer(bool atEof); // aborts or sends eof
 106     size_t putMoreData(const char *buf, size_t size);
 107     bool mayNeedMoreData() const { return !bodySizeKnown() || needsMoreData(); }
 108     bool needsMoreData() const { return bodySizeKnown() && unproducedSize() > 0; }
 109     uint64_t unproducedSize() const; // size of still unproduced data
 110     bool stillProducing(const Producer::Pointer &producer) const { return theProducer == producer; }
 111     void expectProductionEndAfter(uint64_t extraSize); ///< sets or checks body size
 112
 113     // called by consumers
 114     bool setConsumerIfNotLate(const Consumer::Pointer &aConsumer);
 115     void clearConsumer(); // aborts if still piping
 116     size_t getMoreData(MemBuf &buf);
 117     void consume(size_t size);
 118     bool expectMoreAfter(uint64_t offset) const;
 119     bool exhausted() const; // saw eof/abort and all data consumed
 120     bool stillConsuming(const Consumer::Pointer &consumer) const { return theConsumer == consumer; }
 121
 122     // start or continue consuming when there is no consumer
 123     void enableAutoConsumption();
 124
 125     const MemBuf &buf() const { return theBuf; }
 126
 127     const char *status() const; // for debugging only
 128
 129 protected:
 130     // lower-level interface used by Checkout
 131     MemBuf &checkOut(); // obtain raw buffer
 132     void checkIn(Checkout &checkout); // return updated raw buffer
 133     void undoCheckOut(Checkout &checkout); // undo checkout efffect
 134
 135     void scheduleBodyDataNotification();
 136     void scheduleBodyEndNotification();
 137
 138     // keep counters in sync and notify producer or consumer
 139     void postConsume(size_t size);
 140     void postAppend(size_t size);
 141
 142     void startAutoConsumption(); // delayed start of enabled consumption
 143
 144 private:
 145     int64_t  theBodySize;   // expected total content length, if known
 146     Producer::Pointer theProducer; // content producer, if any
 147     Consumer::Pointer theConsumer; // content consumer, if any
 148
 149     uint64_t thePutSize; // ever-increasing total
 150     uint64_t theGetSize; // ever-increasing total
 151
 152     MemBuf theBuf; // produced but not yet consumed content, if any
 153
 154     bool mustAutoConsume; // consume when there is no consumer
 155     bool isCheckedOut; // to keep track of checkout violations
 156
 157     CBDATA_CLASS2(BodyPipe);
 158 };
 159
 160 #endif /* SQUID_BODY_PIPE_H */