src/BodyPipe.h

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