src/adaptation/ecap/XactionRep.cc

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