src/adaptation/ecap/XactionRep.cc

   1 /*
   2  * DEBUG: section 93    eCAP Interface
   3  */
   4 #include "squid.h"
   5 #include <libecap/common/area.h>
   6 #include <libecap/common/delay.h>
   7 #include <libecap/common/named_values.h>
   8 #include <libecap/common/names.h>
   9 #include <libecap/adapter/xaction.h>
  10 #include "adaptation/Answer.h"
  11 #include "adaptation/ecap/Config.h"
  12 #include "adaptation/ecap/XactionRep.h"
  13 #include "adaptation/Initiator.h"
  14 #include "base/TextException.h"
  15 #include "HttpReply.h"
  16 #include "HttpRequest.h"
  17 #include "SquidTime.h"
  18
  19 CBDATA_NAMESPACED_CLASS_INIT(Adaptation::Ecap::XactionRep, XactionRep);
  20
  21 /// a libecap Visitor for converting adapter transaction options to HttpHeader
  22 class OptionsExtractor: public libecap::NamedValueVisitor
  23 {
  24 public:
  25     typedef libecap::Name Name;
  26     typedef libecap::Area Area;
  27
  28     OptionsExtractor(HttpHeader &aMeta): meta(aMeta) {}
  29
  30     // libecap::NamedValueVisitor API
  31     virtual void visit(const Name &name, const Area &value) {
  32         meta.putExt(name.image().c_str(), value.toString().c_str());
  33     }
  34
  35     HttpHeader &meta; ///< where to put extracted options
  36 };
  37
  38 Adaptation::Ecap::XactionRep::XactionRep(
  39     HttpMsg *virginHeader, HttpRequest *virginCause,
  40     const Adaptation::ServicePointer &aService):
  41         AsyncJob("Adaptation::Ecap::XactionRep"),
  42         Adaptation::Initiate("Adaptation::Ecap::XactionRep"),
  43         theService(aService),
  44         theVirginRep(virginHeader), theCauseRep(NULL),
  45         makingVb(opUndecided), proxyingAb(opUndecided),
  46         adaptHistoryId(-1),
  47         vbProductionFinished(false),
  48         abProductionFinished(false), abProductionAtEnd(false)
  49 {
  50     if (virginCause)
  51         theCauseRep = new MessageRep(virginCause);
  52 }
  53
  54 Adaptation::Ecap::XactionRep::~XactionRep()
  55 {
  56     assert(!theMaster);
  57     delete theCauseRep;
  58     theAnswerRep.reset();
  59 }
  60
  61 void
  62 Adaptation::Ecap::XactionRep::master(const AdapterXaction &x)
  63 {
  64     Must(!theMaster);
  65     Must(x != NULL);
  66     theMaster = x;
  67 }
  68
  69 Adaptation::Service &
  70 Adaptation::Ecap::XactionRep::service()
  71 {
  72     Must(theService != NULL);
  73     return *theService;
  74 }
  75
  76 const libecap::Area
  77 Adaptation::Ecap::XactionRep::option(const libecap::Name &name) const
  78 {
  79     if (name == libecap::metaClientIp)
  80         return clientIpValue();
  81     if (name == libecap::metaUserName)
  82         return usernameValue();
  83     if (Adaptation::Config::masterx_shared_name && name == Adaptation::Config::masterx_shared_name)
  84         return masterxSharedValue(name);
  85
  86     // TODO: metaServerIp, metaAuthenticatedUser, and metaAuthenticatedGroups
  87
  88     // If the name is unknown, metaValue returns an emtpy area
  89     return metaValue(name);
  90 }
  91
  92 void
  93 Adaptation::Ecap::XactionRep::visitEachOption(libecap::NamedValueVisitor &visitor) const
  94 {
  95     if (const libecap::Area value = clientIpValue())
  96         visitor.visit(libecap::metaClientIp, value);
  97     if (const libecap::Area value = usernameValue())
  98         visitor.visit(libecap::metaUserName, value);
  99
 100     if (Adaptation::Config::masterx_shared_name) {
 101         const libecap::Name name(Adaptation::Config::masterx_shared_name);
 102         if (const libecap::Area value = masterxSharedValue(name))
 103             visitor.visit(name, value);
 104     }
 105
 106     visitEachMetaHeader(visitor);
 107
 108     // TODO: metaServerIp, metaAuthenticatedUser, and metaAuthenticatedGroups
 109 }
 110
 111 const libecap::Area
 112 Adaptation::Ecap::XactionRep::clientIpValue() const
 113 {
 114     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 115                                  theCauseRep->raw().header : theVirginRep.raw().header);
 116     Must(request);
 117     // TODO: move this logic into HttpRequest::clientIp(bool) and
 118     // HttpRequest::clientIpString(bool) and reuse everywhere
 119     if (TheConfig.send_client_ip && request) {
 120         Ip::Address client_addr;
 121 #if FOLLOW_X_FORWARDED_FOR
 122         if (TheConfig.use_indirect_client) {
 123             client_addr = request->indirect_client_addr;
 124         } else
 125 #endif
 126             client_addr = request->client_addr;
 127         if (!client_addr.isAnyAddr() && !client_addr.isNoAddr()) {
 128             char ntoabuf[MAX_IPSTRLEN] = "";
 129             client_addr.toStr(ntoabuf,MAX_IPSTRLEN);
 130             return libecap::Area::FromTempBuffer(ntoabuf, strlen(ntoabuf));
 131         }
 132     }
 133     return libecap::Area();
 134 }
 135
 136 const libecap::Area
 137 Adaptation::Ecap::XactionRep::usernameValue() const
 138 {
 139 #if USE_AUTH
 140     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 141                                  theCauseRep->raw().header : theVirginRep.raw().header);
 142     Must(request);
 143     if (request->auth_user_request != NULL) {
 144         if (char const *name = request->auth_user_request->username())
 145             return libecap::Area::FromTempBuffer(name, strlen(name));
 146         else if (request->extacl_user.defined() && request->extacl_user.size())
 147             return libecap::Area::FromTempBuffer(request->extacl_user.rawBuf(),
 148                                                  request->extacl_user.size());
 149     }
 150 #endif
 151     return libecap::Area();
 152 }
 153
 154 const libecap::Area
 155 Adaptation::Ecap::XactionRep::masterxSharedValue(const libecap::Name &name) const
 156 {
 157     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 158                                  theCauseRep->raw().header : theVirginRep.raw().header);
 159     Must(request);
 160     if (name.known()) { // must check to avoid empty names matching unset cfg
 161         Adaptation::History::Pointer ah = request->adaptHistory(false);
 162         if (ah != NULL) {
 163             String name, value;
 164             if (ah->getXxRecord(name, value))
 165                 return libecap::Area::FromTempBuffer(value.rawBuf(), value.size());
 166         }
 167     }
 168     return libecap::Area();
 169 }
 170
 171 const libecap::Area
 172 Adaptation::Ecap::XactionRep::metaValue(const libecap::Name &name) const
 173 {
 174     HttpRequest *request = dynamic_cast<HttpRequest*>(theCauseRep ?
 175                            theCauseRep->raw().header : theVirginRep.raw().header);
 176     Must(request);
 177     HttpReply *reply = dynamic_cast<HttpReply*>(theVirginRep.raw().header);
 178
 179     if (name.known()) { // must check to avoid empty names matching unset cfg
 180         typedef Notes::iterator ACAMLI;
 181         for (ACAMLI i = Adaptation::Config::metaHeaders.begin(); i != Adaptation::Config::metaHeaders.end(); ++i) {
 182             if (name == (*i)->key.termedBuf()) {
 183                 if (const char *value = (*i)->match(request, reply))
 184                     return libecap::Area::FromTempString(value);
 185                 else
 186                     return libecap::Area();
 187             }
 188         }
 189     }
 190
 191     return libecap::Area();
 192 }
 193
 194 void
 195 Adaptation::Ecap::XactionRep::visitEachMetaHeader(libecap::NamedValueVisitor &visitor) const
 196 {
 197     HttpRequest *request = dynamic_cast<HttpRequest*>(theCauseRep ?
 198                            theCauseRep->raw().header : theVirginRep.raw().header);
 199     Must(request);
 200     HttpReply *reply = dynamic_cast<HttpReply*>(theVirginRep.raw().header);
 201
 202     typedef Notes::iterator ACAMLI;
 203     for (ACAMLI i = Adaptation::Config::metaHeaders.begin(); i != Adaptation::Config::metaHeaders.end(); ++i) {
 204         const char *v = (*i)->match(request, reply);
 205         if (v) {
 206             const libecap::Name name((*i)->key.termedBuf());
 207             const libecap::Area value = libecap::Area::FromTempString(v);
 208             visitor.visit(name, value);
 209         }
 210     }
 211 }
 212
 213 void
 214 Adaptation::Ecap::XactionRep::start()
 215 {
 216     Must(theMaster);
 217
 218     if (!theVirginRep.raw().body_pipe)
 219         makingVb = opNever; // there is nothing to deliver
 220
 221     HttpRequest *request = dynamic_cast<HttpRequest*> (theCauseRep ?
 222                            theCauseRep->raw().header : theVirginRep.raw().header);
 223     Must(request);
 224
 225     HttpReply *reply = dynamic_cast<HttpReply*>(theVirginRep.raw().header);
 226
 227     Adaptation::History::Pointer ah = request->adaptLogHistory();
 228     if (ah != NULL) {
 229         // retrying=false because ecap never retries transactions
 230         adaptHistoryId = ah->recordXactStart(service().cfg().key, current_time, false);
 231         typedef Notes::iterator ACAMLI;
 232         for (ACAMLI i = Adaptation::Config::metaHeaders.begin(); i != Adaptation::Config::metaHeaders.end(); ++i) {
 233             const char *v = (*i)->match(request, reply);
 234             if (v) {
 235                 if (ah->metaHeaders == NULL)
 236                     ah->metaHeaders = new NotePairs();
 237                 if (!ah->metaHeaders->hasPair((*i)->key.termedBuf(), v))
 238                     ah->metaHeaders->add((*i)->key.termedBuf(), v);
 239             }
 240         }
 241     }
 242
 243     theMaster->start();
 244 }
 245
 246 void
 247 Adaptation::Ecap::XactionRep::swanSong()
 248 {
 249     // clear body_pipes, if any
 250     // this code does not maintain proxying* and canAccessVb states; should it?
 251
 252     if (theAnswerRep != NULL) {
 253         BodyPipe::Pointer body_pipe = answer().body_pipe;
 254         if (body_pipe != NULL) {
 255             Must(body_pipe->stillProducing(this));
 256             stopProducingFor(body_pipe, false);
 257         }
 258     }
 259
 260     BodyPipe::Pointer &body_pipe = theVirginRep.raw().body_pipe;
 261     if (body_pipe != NULL && body_pipe->stillConsuming(this))
 262         stopConsumingFrom(body_pipe);
 263
 264     terminateMaster();
 265
 266     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 267                                  theCauseRep->raw().header : theVirginRep.raw().header);
 268     Must(request);
 269     Adaptation::History::Pointer ah = request->adaptLogHistory();
 270     if (ah != NULL && adaptHistoryId >= 0)
 271         ah->recordXactFinish(adaptHistoryId);
 272
 273     Adaptation::Initiate::swanSong();
 274 }
 275
 276 libecap::Message &
 277 Adaptation::Ecap::XactionRep::virgin()
 278 {
 279     return theVirginRep;
 280 }
 281
 282 const libecap::Message &
 283 Adaptation::Ecap::XactionRep::cause()
 284 {
 285     Must(theCauseRep != NULL);
 286     return *theCauseRep;
 287 }
 288
 289 libecap::Message &
 290 Adaptation::Ecap::XactionRep::adapted()
 291 {
 292     Must(theAnswerRep != NULL);
 293     return *theAnswerRep;
 294 }
 295
 296 Adaptation::Message &
 297 Adaptation::Ecap::XactionRep::answer()
 298 {
 299     MessageRep *rep = dynamic_cast<MessageRep*>(theAnswerRep.get());
 300     Must(rep);
 301     return rep->raw();
 302 }
 303
 304 void
 305 Adaptation::Ecap::XactionRep::terminateMaster()
 306 {
 307     if (theMaster) {
 308         AdapterXaction x = theMaster;
 309         theMaster.reset();
 310         x->stop();
 311     }
 312 }
 313
 314 bool
 315 Adaptation::Ecap::XactionRep::doneAll() const
 316 {
 317     return makingVb >= opComplete && proxyingAb >= opComplete &&
 318            Adaptation::Initiate::doneAll();
 319 }
 320
 321 // stops receiving virgin and enables auto-consumption, dropping any vb bytes
 322 void
 323 Adaptation::Ecap::XactionRep::sinkVb(const char *reason)
 324 {
 325     debugs(93,4, HERE << "sink for " << reason << "; status:" << status());
 326
 327     // we reset raw().body_pipe when we are done, so use this one for checking
 328     const BodyPipePointer &permPipe = theVirginRep.raw().header->body_pipe;
 329     if (permPipe != NULL)
 330         permPipe->enableAutoConsumption();
 331
 332     forgetVb(reason);
 333 }
 334
 335 // stops receiving virgin but preserves it for others to use
 336 void
 337 Adaptation::Ecap::XactionRep::preserveVb(const char *reason)
 338 {
 339     debugs(93,4, HERE << "preserve for " << reason << "; status:" << status());
 340
 341     // we reset raw().body_pipe when we are done, so use this one for checking
 342     const BodyPipePointer &permPipe = theVirginRep.raw().header->body_pipe;
 343     if (permPipe != NULL) {
 344         // if libecap consumed, we cannot preserve
 345         Must(!permPipe->consumedSize());
 346     }
 347
 348     forgetVb(reason);
 349 }
 350
 351 // disassociates us from vb; the last step of sinking or preserving vb
 352 void
 353 Adaptation::Ecap::XactionRep::forgetVb(const char *reason)
 354 {
 355     debugs(93,9, HERE << "forget vb " << reason << "; status:" << status());
 356
 357     BodyPipePointer &p = theVirginRep.raw().body_pipe;
 358     if (p != NULL && p->stillConsuming(this))
 359         stopConsumingFrom(p);
 360
 361     if (makingVb == opUndecided)
 362         makingVb = opNever;
 363     else if (makingVb == opOn)
 364         makingVb = opComplete;
 365 }
 366
 367 void
 368 Adaptation::Ecap::XactionRep::useVirgin()
 369 {
 370     debugs(93,3, HERE << status());
 371     Must(proxyingAb == opUndecided);
 372     proxyingAb = opNever;
 373
 374     preserveVb("useVirgin");
 375
 376     HttpMsg *clone = theVirginRep.raw().header->clone();
 377     // check that clone() copies the pipe so that we do not have to
 378     Must(!theVirginRep.raw().header->body_pipe == !clone->body_pipe);
 379
 380     updateHistory(clone);
 381     sendAnswer(Answer::Forward(clone));
 382     Must(done());
 383 }
 384
 385 void
 386 Adaptation::Ecap::XactionRep::useAdapted(const libecap::shared_ptr<libecap::Message> &m)
 387 {
 388     debugs(93,3, HERE << status());
 389     Must(m);
 390     theAnswerRep = m;
 391     Must(proxyingAb == opUndecided);
 392
 393     HttpMsg *msg = answer().header;
 394     if (!theAnswerRep->body()) { // final, bodyless answer
 395         proxyingAb = opNever;
 396         updateHistory(msg);
 397         sendAnswer(Answer::Forward(msg));
 398     } else { // got answer headers but need to handle body
 399         proxyingAb = opOn;
 400         Must(!msg->body_pipe); // only host can set body pipes
 401         MessageRep *rep = dynamic_cast<MessageRep*>(theAnswerRep.get());
 402         Must(rep);
 403         rep->tieBody(this); // sets us as a producer
 404         Must(msg->body_pipe != NULL); // check tieBody
 405
 406         updateHistory(msg);
 407         sendAnswer(Answer::Forward(msg));
 408
 409         debugs(93,4, HERE << "adapter will produce body" << status());
 410         theMaster->abMake(); // libecap will produce
 411     }
 412 }
 413
 414 void
 415 Adaptation::Ecap::XactionRep::blockVirgin()
 416 {
 417     debugs(93,3, HERE << status());
 418     Must(proxyingAb == opUndecided);
 419     proxyingAb = opNever;
 420
 421     sinkVb("blockVirgin");
 422
 423     updateHistory(NULL);
 424     sendAnswer(Answer::Block(service().cfg().key));
 425     Must(done());
 426 }
 427
 428 /// Called just before sendAnswer() to record adapter meta-information
 429 /// which may affect answer processing and may be needed for logging.
 430 void
 431 Adaptation::Ecap::XactionRep::updateHistory(HttpMsg *adapted)
 432 {
 433     if (!theMaster) // all updates rely on being able to query the adapter
 434         return;
 435
 436     const HttpRequest *request = dynamic_cast<const HttpRequest*>(theCauseRep ?
 437                                  theCauseRep->raw().header : theVirginRep.raw().header);
 438     Must(request);
 439
 440     // TODO: move common ICAP/eCAP logic to Adaptation::Xaction or similar
 441     // TODO: optimize Area-to-String conversion
 442
 443     // update the cross-transactional database if needed
 444     if (const char *xxNameStr = Adaptation::Config::masterx_shared_name) {
 445         Adaptation::History::Pointer ah = request->adaptHistory(true);
 446         if (ah != NULL) {
 447             libecap::Name xxName(xxNameStr); // TODO: optimize?
 448             if (const libecap::Area val = theMaster->option(xxName))
 449                 ah->updateXxRecord(xxNameStr, val.toString().c_str());
 450         }
 451     }
 452
 453     // update the adaptation plan if needed
 454     if (service().cfg().routing) {
 455         String services;
 456         if (const libecap::Area services = theMaster->option(libecap::metaNextServices)) {
 457             Adaptation::History::Pointer ah = request->adaptHistory(true);
 458             if (ah != NULL)
 459                 ah->updateNextServices(services.toString().c_str());
 460         }
 461     } // TODO: else warn (occasionally!) if we got libecap::metaNextServices
 462
 463     // Store received meta headers for adapt::<last_h logformat code use.
 464     // If we already have stored headers from a previous adaptation transaction
 465     // related to the same master transction, they will be replaced.
 466     Adaptation::History::Pointer ah = request->adaptLogHistory();
 467     if (ah != NULL) {
 468         HttpHeader meta(hoReply);
 469         OptionsExtractor extractor(meta);
 470         theMaster->visitEachOption(extractor);
 471         ah->recordMeta(&meta);
 472     }
 473
 474     // Add just-created history to the adapted/cloned request that lacks it.
 475     if (HttpRequest *adaptedReq = dynamic_cast<HttpRequest*>(adapted))
 476         adaptedReq->adaptHistoryImport(*request);
 477 }
 478
 479 void
 480 Adaptation::Ecap::XactionRep::vbDiscard()
 481 {
 482     Must(makingVb == opUndecided);
 483     // if adapter does not need vb, we do not need to send it
 484     sinkVb("vbDiscard");
 485     Must(makingVb == opNever);
 486 }
 487
 488 void
 489 Adaptation::Ecap::XactionRep::vbMake()
 490 {
 491     Must(makingVb == opUndecided);
 492     BodyPipePointer &p = theVirginRep.raw().body_pipe;
 493     Must(p != NULL);
 494     Must(p->setConsumerIfNotLate(this)); // to deliver vb, we must receive vb
 495     makingVb = opOn;
 496 }
 497
 498 void
 499 Adaptation::Ecap::XactionRep::vbStopMaking()
 500 {
 501     Must(makingVb == opOn);
 502     // if adapter does not need vb, we do not need to receive it
 503     sinkVb("vbStopMaking");
 504     Must(makingVb == opComplete);
 505 }
 506
 507 void
 508 Adaptation::Ecap::XactionRep::vbMakeMore()
 509 {
 510     Must(makingVb == opOn); // cannot make more if done proxying
 511     // we cannot guarantee more vb, but we can check that there is a chance
 512     const BodyPipePointer &p = theVirginRep.raw().body_pipe;
 513     Must(p != NULL && p->stillConsuming(this)); // we are plugged in
 514     Must(!p->productionEnded() && p->mayNeedMoreData()); // and may get more
 515 }
 516
 517 libecap::Area
 518 Adaptation::Ecap::XactionRep::vbContent(libecap::size_type o, libecap::size_type s)
 519 {
 520     // We may not be makingVb yet. It should be OK, but see vbContentShift().
 521
 522     const BodyPipePointer &p = theVirginRep.raw().body_pipe;
 523     Must(p != NULL);
 524
 525     // TODO: make MemBuf use size_t?
 526     const size_t haveSize = static_cast<size_t>(p->buf().contentSize());
 527
 528     // convert to Squid types; XXX: check for overflow
 529     const uint64_t offset = static_cast<uint64_t>(o);
 530     Must(offset <= haveSize); // equal iff at the end of content
 531
 532     // nsize means no size limit: all content starting from offset
 533     const size_t size = s == libecap::nsize ?
 534                         haveSize - offset : static_cast<size_t>(s);
 535
 536     // XXX: optimize by making theBody a shared_ptr (see Area::FromTemp*() src)
 537     return libecap::Area::FromTempBuffer(p->buf().content() + offset,
 538                                          min(static_cast<size_t>(haveSize - offset), size));
 539 }
 540
 541 void
 542 Adaptation::Ecap::XactionRep::vbContentShift(libecap::size_type n)
 543 {
 544     // We may not be makingVb yet. It should be OK now, but if BodyPipe
 545     // consume() requirements change, we would have to return empty vbContent
 546     // until the adapter registers as a consumer
 547
 548     BodyPipePointer &p = theVirginRep.raw().body_pipe;
 549     Must(p != NULL);
 550     const size_t size = static_cast<size_t>(n); // XXX: check for overflow
 551     const size_t haveSize = static_cast<size_t>(p->buf().contentSize()); // TODO: make MemBuf use size_t?
 552     p->consume(min(size, haveSize));
 553 }
 554
 555 void
 556 Adaptation::Ecap::XactionRep::noteAbContentDone(bool atEnd)
 557 {
 558     Must(proxyingAb == opOn && !abProductionFinished);
 559     abProductionFinished = true;
 560     abProductionAtEnd = atEnd; // store until ready to stop producing ourselves
 561     debugs(93,5, HERE << "adapted body production ended");
 562     moveAbContent();
 563 }
 564
 565 void
 566 Adaptation::Ecap::XactionRep::noteAbContentAvailable()
 567 {
 568     Must(proxyingAb == opOn && !abProductionFinished);
 569     moveAbContent();
 570 }
 571
 572 #if 0 /* XXX: implement */
 573 void
 574 Adaptation::Ecap::XactionRep::setAdaptedBodySize(const libecap::BodySize &size)
 575 {
 576     Must(answer().body_pipe != NULL);
 577     if (size.known())
 578         answer().body_pipe->setBodySize(size.value());
 579     // else the piped body size is unknown by default
 580 }
 581 #endif
 582
 583 void
 584 Adaptation::Ecap::XactionRep::adaptationDelayed(const libecap::Delay &d)
 585 {
 586     debugs(93,3, HERE << "adapter needs time: " <<
 587            d.state << '/' << d.progress);
 588     // XXX: set timeout?
 589 }
 590
 591 void
 592 Adaptation::Ecap::XactionRep::adaptationAborted()
 593 {
 594     tellQueryAborted(true); // should eCAP support retries?
 595     mustStop("adaptationAborted");
 596 }
 597
 598 bool
 599 Adaptation::Ecap::XactionRep::callable() const
 600 {
 601     return !done();
 602 }
 603
 604 void
 605 Adaptation::Ecap::XactionRep::noteMoreBodySpaceAvailable(RefCount<BodyPipe> bp)
 606 {
 607     Must(proxyingAb == opOn);
 608     moveAbContent();
 609 }
 610
 611 void
 612 Adaptation::Ecap::XactionRep::noteBodyConsumerAborted(RefCount<BodyPipe> bp)
 613 {
 614     Must(proxyingAb == opOn);
 615     stopProducingFor(answer().body_pipe, false);
 616     Must(theMaster);
 617     theMaster->abStopMaking();
 618     proxyingAb = opComplete;
 619 }
 620
 621 void
 622 Adaptation::Ecap::XactionRep::noteMoreBodyDataAvailable(RefCount<BodyPipe> bp)
 623 {
 624     Must(makingVb == opOn); // or we would not be registered as a consumer
 625     Must(theMaster);
 626     theMaster->noteVbContentAvailable();
 627 }
 628
 629 void
 630 Adaptation::Ecap::XactionRep::noteBodyProductionEnded(RefCount<BodyPipe> bp)
 631 {
 632     Must(makingVb == opOn); // or we would not be registered as a consumer
 633     Must(theMaster);
 634     theMaster->noteVbContentDone(true);
 635     vbProductionFinished = true;
 636 }
 637
 638 void
 639 Adaptation::Ecap::XactionRep::noteBodyProducerAborted(RefCount<BodyPipe> bp)
 640 {
 641     Must(makingVb == opOn); // or we would not be registered as a consumer
 642     Must(theMaster);
 643     theMaster->noteVbContentDone(false);
 644     vbProductionFinished = true;
 645 }
 646
 647 void
 648 Adaptation::Ecap::XactionRep::noteInitiatorAborted()
 649 {
 650     mustStop("initiator aborted");
 651 }
 652
 653 // get content from the adapter and put it into the adapted pipe
 654 void
 655 Adaptation::Ecap::XactionRep::moveAbContent()
 656 {
 657     Must(proxyingAb == opOn);
 658     const libecap::Area c = theMaster->abContent(0, libecap::nsize);
 659     debugs(93,5, HERE << "up to " << c.size << " bytes");
 660     if (c.size == 0 && abProductionFinished) { // no ab now and in the future
 661         stopProducingFor(answer().body_pipe, abProductionAtEnd);
 662         proxyingAb = opComplete;
 663         debugs(93,5, HERE << "last adapted body data retrieved");
 664     } else if (c.size > 0) {
 665         if (const size_t used = answer().body_pipe->putMoreData(c.start, c.size))
 666             theMaster->abContentShift(used);
 667     }
 668 }
 669
 670 const char *
 671 Adaptation::Ecap::XactionRep::status() const
 672 {
 673     static MemBuf buf;
 674     buf.reset();
 675
 676     buf.append(" [", 2);
 677
 678     if (makingVb)
 679         buf.Printf("M%d", static_cast<int>(makingVb));
 680
 681     const BodyPipePointer &vp = theVirginRep.raw().body_pipe;
 682     if (!vp)
 683         buf.append(" !V", 3);
 684     else if (vp->stillConsuming(const_cast<XactionRep*>(this)))
 685         buf.append(" Vc", 3);
 686     else
 687         buf.append(" V?", 3);
 688
 689     if (vbProductionFinished)
 690         buf.append(".", 1);
 691
 692     buf.Printf(" A%d", static_cast<int>(proxyingAb));
 693
 694     if (proxyingAb == opOn) {
 695         MessageRep *rep = dynamic_cast<MessageRep*>(theAnswerRep.get());
 696         Must(rep);
 697         const BodyPipePointer &ap = rep->raw().body_pipe;
 698         if (!ap)
 699             buf.append(" !A", 3);
 700         else if (ap->stillProducing(const_cast<XactionRep*>(this)))
 701             buf.append(" Ap", 3);
 702         else
 703             buf.append(" A?", 3);
 704     }
 705
 706     buf.Printf(" %s%u]", id.Prefix, id.value);
 707
 708     buf.terminate();
 709
 710     return buf.content();
 711 }