]>
Commit | Line | Data |
---|---|---|
1b39caaa | 1 | #ifndef SQUID_BODY_PIPE_H |
2 | #define SQUID_BODY_PIPE_H | |
3 | ||
d1e045c3 | 4 | #include "base/AsyncJob.h" |
4299f876 | 5 | #include "base/CbcPointer.h" |
602d9612 | 6 | #include "MemBuf.h" |
1b39caaa | 7 | |
8 | class BodyPipe; | |
9 | ||
d1c37eab AJ |
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 | */ | |
26ac0430 AJ |
14 | class BodyProducer: virtual public AsyncJob |
15 | { | |
16 | public: | |
4299f876 AR |
17 | typedef CbcPointer<BodyProducer> Pointer; |
18 | ||
26ac0430 AJ |
19 | BodyProducer():AsyncJob("BodyProducer") {} |
20 | virtual ~BodyProducer() {} | |
1b39caaa | 21 | |
26ac0430 AJ |
22 | virtual void noteMoreBodySpaceAvailable(RefCount<BodyPipe> bp) = 0; |
23 | virtual void noteBodyConsumerAborted(RefCount<BodyPipe> bp) = 0; | |
1b39caaa | 24 | |
26ac0430 AJ |
25 | protected: |
26 | void stopProducingFor(RefCount<BodyPipe> &pipe, bool atEof); | |
1b39caaa | 27 | }; |
28 | ||
d1c37eab AJ |
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 | */ | |
26ac0430 AJ |
34 | class BodyConsumer: virtual public AsyncJob |
35 | { | |
36 | public: | |
4299f876 AR |
37 | typedef CbcPointer<BodyConsumer> Pointer; |
38 | ||
26ac0430 AJ |
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); | |
1b39caaa | 48 | }; |
49 | ||
d1c37eab AJ |
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 | */ | |
26ac0430 AJ |
53 | class BodyPipeCheckout |
54 | { | |
55 | public: | |
56 | friend class BodyPipe; | |
1b39caaa | 57 | |
26ac0430 AJ |
58 | public: |
59 | BodyPipeCheckout(BodyPipe &pipe); // checks out | |
60 | ~BodyPipeCheckout(); // aborts checkout unless checkedIn | |
1b39caaa | 61 | |
26ac0430 | 62 | void checkIn(); |
1b39caaa | 63 | |
26ac0430 AJ |
64 | public: |
65 | BodyPipe &pipe; | |
66 | MemBuf &buf; | |
67 | const uint64_t offset; // of current content, relative to the body start | |
1b39caaa | 68 | |
26ac0430 AJ |
69 | protected: |
70 | const size_t checkedOutSize; | |
71 | bool checkedIn; | |
1b39caaa | 72 | |
26ac0430 AJ |
73 | private: |
74 | BodyPipeCheckout(const BodyPipeCheckout &); // prevent copying | |
75 | BodyPipeCheckout &operator =(const BodyPipeCheckout &); // prevent assignment | |
1b39caaa | 76 | }; |
77 | ||
d1c37eab AJ |
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 | */ | |
26ac0430 AJ |
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; | |
1b39caaa | 89 | |
4290445a | 90 | enum { MaxCapacity = 64*1024 }; |
1b39caaa | 91 | |
26ac0430 | 92 | friend class BodyPipeCheckout; |
1b39caaa | 93 | |
26ac0430 AJ |
94 | public: |
95 | BodyPipe(Producer *aProducer); | |
96 | ~BodyPipe(); // asserts that producer and consumer are cleared | |
1b39caaa | 97 | |
26ac0430 AJ |
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; } | |
dc0017a9 | 102 | uint64_t producedSize() const { return thePutSize; } |
26ac0430 | 103 | bool productionEnded() const { return !theProducer; } |
1b39caaa | 104 | |
26ac0430 AJ |
105 | // called by producers |
106 | void clearProducer(bool atEof); // aborts or sends eof | |
107 | size_t putMoreData(const char *buf, size_t size); | |
108 | bool mayNeedMoreData() const { return !bodySizeKnown() || needsMoreData(); } | |
109 | bool needsMoreData() const { return bodySizeKnown() && unproducedSize() > 0; } | |
110 | uint64_t unproducedSize() const; // size of still unproduced data | |
4299f876 | 111 | bool stillProducing(const Producer::Pointer &producer) const { return theProducer == producer; } |
83c51da9 | 112 | void expectProductionEndAfter(uint64_t extraSize); ///< sets or checks body size |
1b39caaa | 113 | |
26ac0430 | 114 | // called by consumers |
4299f876 | 115 | bool setConsumerIfNotLate(const Consumer::Pointer &aConsumer); |
26ac0430 | 116 | void clearConsumer(); // aborts if still piping |
abe286b8 | 117 | void expectNoConsumption(); ///< there will be no more setConsumer() calls |
26ac0430 AJ |
118 | size_t getMoreData(MemBuf &buf); |
119 | void consume(size_t size); | |
120 | bool expectMoreAfter(uint64_t offset) const; | |
121 | bool exhausted() const; // saw eof/abort and all data consumed | |
4299f876 | 122 | bool stillConsuming(const Consumer::Pointer &consumer) const { return theConsumer == consumer; } |
1b39caaa | 123 | |
26ac0430 AJ |
124 | // start or continue consuming when there is no consumer |
125 | void enableAutoConsumption(); | |
1b39caaa | 126 | |
26ac0430 | 127 | const MemBuf &buf() const { return theBuf; } |
1b39caaa | 128 | |
26ac0430 | 129 | const char *status() const; // for debugging only |
1b39caaa | 130 | |
26ac0430 AJ |
131 | protected: |
132 | // lower-level interface used by Checkout | |
133 | MemBuf &checkOut(); // obtain raw buffer | |
134 | void checkIn(Checkout &checkout); // return updated raw buffer | |
135 | void undoCheckOut(Checkout &checkout); // undo checkout efffect | |
1b39caaa | 136 | |
26ac0430 AJ |
137 | void scheduleBodyDataNotification(); |
138 | void scheduleBodyEndNotification(); | |
1b39caaa | 139 | |
26ac0430 AJ |
140 | // keep counters in sync and notify producer or consumer |
141 | void postConsume(size_t size); | |
142 | void postAppend(size_t size); | |
1b39caaa | 143 | |
26ac0430 | 144 | void startAutoConsumption(); // delayed start of enabled consumption |
5121d48e | 145 | |
26ac0430 AJ |
146 | private: |
147 | int64_t theBodySize; // expected total content length, if known | |
4299f876 AR |
148 | Producer::Pointer theProducer; // content producer, if any |
149 | Consumer::Pointer theConsumer; // content consumer, if any | |
1b39caaa | 150 | |
26ac0430 AJ |
151 | uint64_t thePutSize; // ever-increasing total |
152 | uint64_t theGetSize; // ever-increasing total | |
1b39caaa | 153 | |
26ac0430 | 154 | MemBuf theBuf; // produced but not yet consumed content, if any |
1b39caaa | 155 | |
26ac0430 | 156 | bool mustAutoConsume; // consume when there is no consumer |
abe286b8 | 157 | bool abortedConsumption; ///< called BodyProducer::noteBodyConsumerAborted |
26ac0430 | 158 | bool isCheckedOut; // to keep track of checkout violations |
1b39caaa | 159 | |
26ac0430 | 160 | CBDATA_CLASS2(BodyPipe); |
1b39caaa | 161 | }; |
162 | ||
163 | #endif /* SQUID_BODY_PIPE_H */ |