[apt.git] / apt-pkg / acquire-worker.cc

// -*- mode: cpp; mode: fold -*-
// Description								/*{{{*/
// $Id: acquire-worker.cc,v 1.34 2001/05/22 04:42:54 jgg Exp $
/* ######################################################################

   Acquire Worker 

   The worker process can startup either as a Configuration prober
   or as a queue runner. As a configuration prober it only reads the
   configuration message and 
   
   ##################################################################### */
									/*}}}*/
// Include Files							/*{{{*/
#include <config.h>

#include <apt-pkg/acquire.h>
#include <apt-pkg/acquire-worker.h>
#include <apt-pkg/acquire-item.h>
#include <apt-pkg/configuration.h>
#include <apt-pkg/error.h>
#include <apt-pkg/fileutl.h>
#include <apt-pkg/strutl.h>
#include <apt-pkg/hashes.h>

#include <string>
#include <vector>
#include <iostream>
#include <sstream>

#include <sys/stat.h>
#include <stdlib.h>
#include <unistd.h>
#include <signal.h>
#include <stdio.h>
#include <errno.h>

#include <apti18n.h>
									/*}}}*/

using namespace std;

// Worker::Worker - Constructor for Queue startup			/*{{{*/
// ---------------------------------------------------------------------
/* */
pkgAcquire::Worker::Worker(Queue *Q,MethodConfig *Cnf,
			   pkgAcquireStatus *Log) : Log(Log)
{
   OwnerQ = Q;
   Config = Cnf;
   Access = Cnf->Access;
   CurrentItem = 0;
   TotalSize = 0;
   CurrentSize = 0;
   
   Construct();   
}
									/*}}}*/
// Worker::Worker - Constructor for method config startup		/*{{{*/
// ---------------------------------------------------------------------
/* */
pkgAcquire::Worker::Worker(MethodConfig *Cnf)
{
   OwnerQ = 0;
   Config = Cnf;
   Access = Cnf->Access;
   CurrentItem = 0;
   TotalSize = 0;
   CurrentSize = 0;
   
   Construct();   
}
									/*}}}*/
// Worker::Construct - Constructor helper				/*{{{*/
// ---------------------------------------------------------------------
/* */
void pkgAcquire::Worker::Construct()
{
   NextQueue = 0;
   NextAcquire = 0;
   Process = -1;
   InFd = -1;
   OutFd = -1;
   OutReady = false;
   InReady = false;
   Debug = _config->FindB("Debug::pkgAcquire::Worker",false);
}
									/*}}}*/
// Worker::~Worker - Destructor						/*{{{*/
// ---------------------------------------------------------------------
/* */
pkgAcquire::Worker::~Worker()
{
   close(InFd);
   close(OutFd);
   
   if (Process > 0)
   {
      /* Closing of stdin is the signal to exit and die when the process
         indicates it needs cleanup */
      if (Config->NeedsCleanup == false)
	 kill(Process,SIGINT);
      ExecWait(Process,Access.c_str(),true);
   }   
}
									/*}}}*/
// Worker::Start - Start the worker process				/*{{{*/
// ---------------------------------------------------------------------
/* This forks the method and inits the communication channel */
bool pkgAcquire::Worker::Start()
{
   // Get the method path
   string Method = _config->FindDir("Dir::Bin::Methods") + Access;
   if (FileExists(Method) == false)
   {
      _error->Error(_("The method driver %s could not be found."),Method.c_str());
      if (Access == "https")
	 _error->Notice(_("Is the package %s installed?"), "apt-transport-https");
      return false;
   }

   if (Debug == true)
      clog << "Starting method '" << Method << '\'' << endl;

   // Create the pipes
   int Pipes[4] = {-1,-1,-1,-1};
   if (pipe(Pipes) != 0 || pipe(Pipes+2) != 0)
   {
      _error->Errno("pipe","Failed to create IPC pipe to subprocess");
      for (int I = 0; I != 4; I++)
	 close(Pipes[I]);
      return false;
   }
   for (int I = 0; I != 4; I++)
      SetCloseExec(Pipes[I],true);
   
   // Fork off the process
   Process = ExecFork();
   if (Process == 0)
   {
      // Setup the FDs
      dup2(Pipes[1],STDOUT_FILENO);
      dup2(Pipes[2],STDIN_FILENO);
      SetCloseExec(STDOUT_FILENO,false);
      SetCloseExec(STDIN_FILENO,false);      
      SetCloseExec(STDERR_FILENO,false);
      
      const char *Args[2];
      Args[0] = Method.c_str();
      Args[1] = 0;
      execv(Args[0],(char **)Args);
      cerr << "Failed to exec method " << Args[0] << endl;
      _exit(100);
   }

   // Fix up our FDs
   InFd = Pipes[0];
   OutFd = Pipes[3];
   SetNonBlock(Pipes[0],true);
   SetNonBlock(Pipes[3],true);
   close(Pipes[1]);
   close(Pipes[2]);
   OutReady = false;
   InReady = true;
   
   // Read the configuration data
   if (WaitFd(InFd) == false ||
       ReadMessages() == false)
      return _error->Error(_("Method %s did not start correctly"),Method.c_str());

   RunMessages();
   if (OwnerQ != 0)
      SendConfiguration();
   
   return true;
}
									/*}}}*/
// Worker::ReadMessages - Read all pending messages into the list	/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::ReadMessages()
{
   if (::ReadMessages(InFd,MessageQueue) == false)
      return MethodFailure();
   return true;
}
									/*}}}*/
// Worker::RunMessage - Empty the message queue				/*{{{*/
// ---------------------------------------------------------------------
/* This takes the messages from the message queue and runs them through
   the parsers in order. */
bool pkgAcquire::Worker::RunMessages()
{
   while (MessageQueue.empty() == false)
   {
      string Message = MessageQueue.front();
      MessageQueue.erase(MessageQueue.begin());

      if (Debug == true)
	 clog << " <- " << Access << ':' << QuoteString(Message,"\n") << endl;
      
      // Fetch the message number
      char *End;
      int Number = strtol(Message.c_str(),&End,10);
      if (End == Message.c_str())
	 return _error->Error("Invalid message from method %s: %s",Access.c_str(),Message.c_str());

      string URI = LookupTag(Message,"URI");
      pkgAcquire::Queue::QItem *Itm = 0;
      if (URI.empty() == false)
	 Itm = OwnerQ->FindItem(URI,this);

      // update used mirror
      string UsedMirror = LookupTag(Message,"UsedMirror", "");
      if (!UsedMirror.empty() && 
          Itm && 
          Itm->Description.find(" ") != string::npos) 
      {
         Itm->Description.replace(0, Itm->Description.find(" "), UsedMirror);
         // FIXME: will we need this as well?
         //Itm->ShortDesc = UsedMirror;
      }
      
      // Determine the message number and dispatch
      switch (Number)
      {
	 // 100 Capabilities
	 case 100:
	 if (Capabilities(Message) == false)
	    return _error->Error("Unable to process Capabilities message from %s",Access.c_str());
	 break;
	 
	 // 101 Log
	 case 101:
	 if (Debug == true)
	    clog << " <- (log) " << LookupTag(Message,"Message") << endl;
	 break;
	 
	 // 102 Status
	 case 102:
	 Status = LookupTag(Message,"Message");
	 break;
	    
         // 103 Redirect
         case 103:
         {
            if (Itm == 0)
            {
               _error->Error("Method gave invalid 103 Redirect message");
               break;
            }
 
            string NewURI = LookupTag(Message,"New-URI",URI.c_str());
            Itm->URI = NewURI;

	    ItemDone();

	    pkgAcquire::Item *Owner = Itm->Owner;
	    pkgAcquire::ItemDesc Desc = *Itm;

	    // Change the status so that it can be dequeued
	    Owner->Status = pkgAcquire::Item::StatIdle;
	    // Mark the item as done (taking care of all queues)
	    // and then put it in the main queue again
	    OwnerQ->ItemDone(Itm);
	    OwnerQ->Owner->Enqueue(Desc);

	    if (Log != 0)
	       Log->Done(Desc);
            break;
         }
   
	 // 200 URI Start
	 case 200:
	 {
	    if (Itm == 0)
	    {
	       _error->Error("Method gave invalid 200 URI Start message");
	       break;
	    }
	    
	    CurrentItem = Itm;
	    CurrentSize = 0;
	    TotalSize = strtoull(LookupTag(Message,"Size","0").c_str(), NULL, 10);
	    ResumePoint = strtoull(LookupTag(Message,"Resume-Point","0").c_str(), NULL, 10);
	    Itm->Owner->Start(Message,strtoull(LookupTag(Message,"Size","0").c_str(), NULL, 10));

	    // Display update before completion
	    if (Log != 0 && Log->MorePulses == true)
	       Log->Pulse(Itm->Owner->GetOwner());
	    
	    if (Log != 0)
	       Log->Fetch(*Itm);

	    break;
	 }
	 
	 // 201 URI Done
	 case 201:
	 {
	    if (Itm == 0)
	    {
	       _error->Error("Method gave invalid 201 URI Done message");
	       break;
	    }
	    
	    pkgAcquire::Item *Owner = Itm->Owner;
	    pkgAcquire::ItemDesc Desc = *Itm;
	    
	    // Display update before completion
	    if (Log != 0 && Log->MorePulses == true)
	       Log->Pulse(Owner->GetOwner());
	    
	    OwnerQ->ItemDone(Itm);
	    unsigned long long const ServerSize = strtoull(LookupTag(Message,"Size","0").c_str(), NULL, 10);
            bool isHit = StringToBool(LookupTag(Message,"IMS-Hit"),false) ||
                         StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false);
            // Using the https method the server might return 200, but the
            // If-Modified-Since condition is not satsified, libcurl will
            // discard the download. In this case, however, TotalSize will be
            // set to the actual size of the file, while ServerSize will be set
            // to 0. Therefore, if the item is marked as a hit and the
            // downloaded size (ServerSize) is 0, we ignore TotalSize.
	    if (TotalSize != 0 && (!isHit || ServerSize != 0) && ServerSize != TotalSize)
	       _error->Warning("Size of file %s is not what the server reported %s %llu",
			       Owner->DestFile.c_str(), LookupTag(Message,"Size","0").c_str(),TotalSize);

	    // see if there is a hash to verify
	    HashStringList RecivedHashes;
	    HashStringList expectedHashes = Owner->HashSums();
	    for (HashStringList::const_iterator hs = expectedHashes.begin(); hs != expectedHashes.end(); ++hs)
	    {
	       std::string const tagname = hs->HashType() + "-Hash";
	       std::string const hashsum = LookupTag(Message, tagname.c_str());
	       if (hashsum.empty() == false)
		  RecivedHashes.push_back(HashString(hs->HashType(), hashsum));
	    }

	    if(_config->FindB("Debug::pkgAcquire::Auth", false) == true)
	    {
	       std::clog << "201 URI Done: " << Owner->DescURI() << endl
		  << "RecivedHash:" << endl;
	       for (HashStringList::const_iterator hs = RecivedHashes.begin(); hs != RecivedHashes.end(); ++hs)
		  std::clog <<  "\t- " << hs->toStr() << std::endl;
	       std::clog << "ExpectedHash:" << endl;
	       for (HashStringList::const_iterator hs = expectedHashes.begin(); hs != expectedHashes.end(); ++hs)
		  std::clog <<  "\t- " << hs->toStr() << std::endl;
	       std::clog << endl;
	    }
	    Owner->Done(Message, ServerSize, RecivedHashes, Config);
	    ItemDone();

	    // Log that we are done
	    if (Log != 0)
	    {
	       if (isHit)
	       {
		  /* Hide 'hits' for local only sources - we also manage to
		     hide gets */
		  if (Config->LocalOnly == false)
		     Log->IMSHit(Desc);
	       }	       
	       else
		  Log->Done(Desc);
	    }
	    break;
	 }	 
	 
	 // 400 URI Failure
	 case 400:
	 {
	    if (Itm == 0)
	    {
	       _error->Error("Method gave invalid 400 URI Failure message");
	       break;
	    }

	    // Display update before completion
	    if (Log != 0 && Log->MorePulses == true)
	       Log->Pulse(Itm->Owner->GetOwner());
	    
	    pkgAcquire::Item *Owner = Itm->Owner;
	    pkgAcquire::ItemDesc Desc = *Itm;
	    OwnerQ->ItemDone(Itm);

	    // set some status
	    if(LookupTag(Message,"FailReason") == "Timeout" || 
	       LookupTag(Message,"FailReason") == "TmpResolveFailure" ||
	       LookupTag(Message,"FailReason") == "ResolveFailure" ||
	       LookupTag(Message,"FailReason") == "ConnectionRefused") 
	       Owner->Status = pkgAcquire::Item::StatTransientNetworkError;

	    Owner->Failed(Message,Config);
	    ItemDone();

	    if (Log != 0)
	       Log->Fail(Desc);

	    break;
	 }	 
	 
	 // 401 General Failure
	 case 401:
	 _error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
	 break;
	 
	 // 403 Media Change
	 case 403:
	 MediaChange(Message); 
	 break;
      }      
   }
   return true;
}
									/*}}}*/
// Worker::Capabilities - 100 Capabilities handler			/*{{{*/
// ---------------------------------------------------------------------
/* This parses the capabilities message and dumps it into the configuration
   structure. */
bool pkgAcquire::Worker::Capabilities(string Message)
{
   if (Config == 0)
      return true;
   
   Config->Version = LookupTag(Message,"Version");
   Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
   Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
   Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
   Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
   Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false);
   Config->Removable = StringToBool(LookupTag(Message,"Removable"),false);

   // Some debug text
   if (Debug == true)
   {
      clog << "Configured access method " << Config->Access << endl;
      clog << "Version:" << Config->Version <<
	      " SingleInstance:" << Config->SingleInstance <<
	      " Pipeline:" << Config->Pipeline << 
	      " SendConfig:" << Config->SendConfig << 
	      " LocalOnly: " << Config->LocalOnly << 
	      " NeedsCleanup: " << Config->NeedsCleanup << 
	      " Removable: " << Config->Removable << endl;
   }
   
   return true;
}
									/*}}}*/
// Worker::MediaChange - Request a media change				/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::MediaChange(string Message)
{
   int status_fd = _config->FindI("APT::Status-Fd",-1);
   if(status_fd > 0) 
   {
      string Media = LookupTag(Message,"Media");
      string Drive = LookupTag(Message,"Drive"); 
      ostringstream msg,status;
      ioprintf(msg,_("Please insert the disc labeled: "
		     "'%s' "
		     "in the drive '%s' and press enter."),
	       Media.c_str(),Drive.c_str());
      status << "media-change: "  // message
	     << Media  << ":"     // media
	     << Drive  << ":"     // drive
	     << msg.str()         // l10n message
	     << endl;

      std::string const dlstatus = status.str();
      FileFd::Write(status_fd, dlstatus.c_str(), dlstatus.size());
   }

   if (Log == 0 || Log->MediaChange(LookupTag(Message,"Media"),
				    LookupTag(Message,"Drive")) == false)
   {
      char S[300];
      snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
      if (Debug == true)
	 clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
      OutQueue += S;
      OutReady = true;
      return true;
   }

   char S[300];
   snprintf(S,sizeof(S),"603 Media Changed\n\n");
   if (Debug == true)
      clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
   OutQueue += S;
   OutReady = true;
   return true;
}
									/*}}}*/
// Worker::SendConfiguration - Send the config to the method		/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::SendConfiguration()
{
   if (Config->SendConfig == false)
      return true;

   if (OutFd == -1)
      return false;

   /* Write out all of the configuration directives by walking the
      configuration tree */
   std::ostringstream Message;
   Message << "601 Configuration\n";
   _config->Dump(Message, NULL, "Config-Item: %F=%V\n", false);
   Message << '\n';

   if (Debug == true)
      clog << " -> " << Access << ':' << QuoteString(Message.str(),"\n") << endl;
   OutQueue += Message.str();
   OutReady = true;

   return true;
}
									/*}}}*/
// Worker::QueueItem - Add an item to the outbound queue		/*{{{*/
// ---------------------------------------------------------------------
/* Send a URI Acquire message to the method */
bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
{
   if (OutFd == -1)
      return false;
   
   string Message = "600 URI Acquire\n";
   Message.reserve(300);
   Message += "URI: " + Item->URI;
   Message += "\nFilename: " + Item->Owner->DestFile;
   HashStringList const hsl = Item->Owner->HashSums();
   for (HashStringList::const_iterator hs = hsl.begin(); hs != hsl.end(); ++hs)
      Message += "\nExpected-" + hs->HashType() + ": " + hs->HashValue();
   Message += Item->Owner->Custom600Headers();
   Message += "\n\n";
   
   if (Debug == true)
      clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
   OutQueue += Message;
   OutReady = true;
   
   return true;
}
									/*}}}*/
// Worker::OutFdRead - Out bound FD is ready				/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::OutFdReady()
{
   int Res;
   do
   {
      Res = write(OutFd,OutQueue.c_str(),OutQueue.length());
   }
   while (Res < 0 && errno == EINTR);

   if (Res <= 0)
      return MethodFailure();

   OutQueue.erase(0,Res);
   if (OutQueue.empty() == true)
      OutReady = false;
   
   return true;
}
									/*}}}*/
// Worker::InFdRead - In bound FD is ready				/*{{{*/
// ---------------------------------------------------------------------
/* */
bool pkgAcquire::Worker::InFdReady()
{
   if (ReadMessages() == false)
      return false;
   RunMessages();
   return true;
}
									/*}}}*/
// Worker::MethodFailure - Called when the method fails			/*{{{*/
// ---------------------------------------------------------------------
/* This is called when the method is believed to have failed, probably because
   read returned -1. */
bool pkgAcquire::Worker::MethodFailure()
{
   _error->Error("Method %s has died unexpectedly!",Access.c_str());
   
   // do not reap the child here to show meaningfull error to the user
   ExecWait(Process,Access.c_str(),false);
   Process = -1;
   close(InFd);
   close(OutFd);
   InFd = -1;
   OutFd = -1;
   OutReady = false;
   InReady = false;
   OutQueue = string();
   MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
   
   return false;
}
									/*}}}*/
// Worker::Pulse - Called periodically 					/*{{{*/
// ---------------------------------------------------------------------
/* */
void pkgAcquire::Worker::Pulse()
{
   if (CurrentItem == 0)
      return;
 
   struct stat Buf;
   if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
      return;
   CurrentSize = Buf.st_size;
   
   // Hmm? Should not happen...
   if (CurrentSize > TotalSize && TotalSize != 0)
      TotalSize = CurrentSize;
}
									/*}}}*/
// Worker::ItemDone - Called when the current item is finished		/*{{{*/
// ---------------------------------------------------------------------
/* */
void pkgAcquire::Worker::ItemDone()
{
   CurrentItem = 0;
   CurrentSize = 0;
   TotalSize = 0;
   Status = string();
}
									/*}}}*/
Commit	Line	Data
0118833a AL	1	// -- mode: cpp; mode: fold --
0118833a AL	2	// Description /{{{/
24a0d63a	3	// $Id: acquire-worker.cc,v 1.34 2001/05/22 04:42:54 jgg Exp $
0118833a AL	4	/* ######################################################################
	5
	6	Acquire Worker
	7
3b5421b4 AL	8	The worker process can startup either as a Configuration prober
	9	or as a queue runner. As a configuration prober it only reads the
	10	configuration message and
	11
0118833a AL	12	##################################################################### */
	13	/}}}/
	14	// Include Files /{{{/
ea542140 DK	15	#include <config.h>
ea542140 DK	16
453b82a3	17	#include <apt-pkg/acquire.h>
0118833a	18	#include <apt-pkg/acquire-worker.h>
0a8a80e5	19	#include <apt-pkg/acquire-item.h>
3b5421b4 AL	20	#include <apt-pkg/configuration.h>
	21	#include <apt-pkg/error.h>
	22	#include <apt-pkg/fileutl.h>
cdcc6d34	23	#include <apt-pkg/strutl.h>
453b82a3	24	#include <apt-pkg/hashes.h>
3b5421b4	25
453b82a3 DK	26	#include <string>
453b82a3 DK	27	#include <vector>
24a0d63a	28	#include <iostream>
80a26ed1	29	#include <sstream>
ea542140	30
8267fe24	31	#include <sys/stat.h>
453b82a3	32	#include <stdlib.h>
3b5421b4 AL	33	#include <unistd.h>
3b5421b4 AL	34	#include <signal.h>
542ec555	35	#include <stdio.h>
b0db36b1	36	#include <errno.h>
ea542140 DK	37
ea542140 DK	38	#include <apti18n.h>
3b5421b4 AL	39	/}}}/
3b5421b4 AL	40
24a0d63a AL	41	using namespace std;
24a0d63a AL	42
3b5421b4 AL	43	// Worker::Worker - Constructor for Queue startup /{{{/
	44	// ---------------------------------------------------------------------
	45	/* */
8267fe24 AL	46	pkgAcquire::Worker::Worker(Queue Q,MethodConfig Cnf,
8267fe24 AL	47	pkgAcquireStatus *Log) : Log(Log)
3b5421b4 AL	48	{
3b5421b4 AL	49	OwnerQ = Q;
0a8a80e5 AL	50	Config = Cnf;
	51	Access = Cnf->Access;
	52	CurrentItem = 0;
18ef0a78 AL	53	TotalSize = 0;
	54	CurrentSize = 0;
	55
3b5421b4 AL	56	Construct();
	57	}
	58	/}}}/
	59	// Worker::Worker - Constructor for method config startup /{{{/
	60	// ---------------------------------------------------------------------
	61	/* */
	62	pkgAcquire::Worker::Worker(MethodConfig *Cnf)
	63	{
	64	OwnerQ = 0;
	65	Config = Cnf;
	66	Access = Cnf->Access;
0a8a80e5	67	CurrentItem = 0;
18ef0a78 AL	68	TotalSize = 0;
18ef0a78 AL	69	CurrentSize = 0;
0a8a80e5	70
3b5421b4 AL	71	Construct();
	72	}
	73	/}}}/
	74	// Worker::Construct - Constructor helper /{{{/
	75	// ---------------------------------------------------------------------
	76	/* */
	77	void pkgAcquire::Worker::Construct()
	78	{
0a8a80e5 AL	79	NextQueue = 0;
0a8a80e5 AL	80	NextAcquire = 0;
3b5421b4 AL	81	Process = -1;
	82	InFd = -1;
	83	OutFd = -1;
0a8a80e5 AL	84	OutReady = false;
0a8a80e5 AL	85	InReady = false;
3b5421b4 AL	86	Debug = _config->FindB("Debug::pkgAcquire::Worker",false);
	87	}
	88	/}}}/
	89	// Worker::~Worker - Destructor /{{{/
	90	// ---------------------------------------------------------------------
	91	/* */
	92	pkgAcquire::Worker::~Worker()
	93	{
	94	close(InFd);
	95	close(OutFd);
	96
	97	if (Process > 0)
0a8a80e5	98	{
8e5fc8f5 AL	99	/* Closing of stdin is the signal to exit and die when the process
	100	indicates it needs cleanup */
	101	if (Config->NeedsCleanup == false)
	102	kill(Process,SIGINT);
ddc1d8d0	103	ExecWait(Process,Access.c_str(),true);
0a8a80e5	104	}
3b5421b4 AL	105	}
	106	/}}}/
	107	// Worker::Start - Start the worker process /{{{/
	108	// ---------------------------------------------------------------------
	109	/* This forks the method and inits the communication channel */
	110	bool pkgAcquire::Worker::Start()
	111	{
	112	// Get the method path
	113	string Method = _config->FindDir("Dir::Bin::Methods") + Access;
	114	if (FileExists(Method) == false)
9082a1fc DK	115	{
	116	_error->Error(_("The method driver %s could not be found."),Method.c_str());
	117	if (Access == "https")
	118	_error->Notice(_("Is the package %s installed?"), "apt-transport-https");
	119	return false;
	120	}
3b5421b4 AL	121
	122	if (Debug == true)
	123	clog << "Starting method '" << Method << '\'' << endl;
	124
	125	// Create the pipes
	126	int Pipes[4] = {-1,-1,-1,-1};
	127	if (pipe(Pipes) != 0 \|\| pipe(Pipes+2) != 0)
	128	{
	129	_error->Errno("pipe","Failed to create IPC pipe to subprocess");
	130	for (int I = 0; I != 4; I++)
	131	close(Pipes[I]);
	132	return false;
	133	}
8b89e57f	134	for (int I = 0; I != 4; I++)
4490f2de	135	SetCloseExec(Pipes[I],true);
8b89e57f	136
3b5421b4	137	// Fork off the process
54676e1a	138	Process = ExecFork();
3b5421b4 AL	139	if (Process == 0)
	140	{
	141	// Setup the FDs
	142	dup2(Pipes[1],STDOUT_FILENO);
	143	dup2(Pipes[2],STDIN_FILENO);
3b5421b4 AL	144	SetCloseExec(STDOUT_FILENO,false);
	145	SetCloseExec(STDIN_FILENO,false);
	146	SetCloseExec(STDERR_FILENO,false);
	147
	148	const char *Args[2];
	149	Args[0] = Method.c_str();
	150	Args[1] = 0;
	151	execv(Args[0],(char **)Args);
	152	cerr << "Failed to exec method " << Args[0] << endl;
0dbb95d8	153	_exit(100);
3b5421b4 AL	154	}
	155
	156	// Fix up our FDs
	157	InFd = Pipes[0];
	158	OutFd = Pipes[3];
	159	SetNonBlock(Pipes[0],true);
	160	SetNonBlock(Pipes[3],true);
	161	close(Pipes[1]);
	162	close(Pipes[2]);
0a8a80e5 AL	163	OutReady = false;
0a8a80e5 AL	164	InReady = true;
3b5421b4 AL	165
	166	// Read the configuration data
	167	if (WaitFd(InFd) == false \|\|
	168	ReadMessages() == false)
b2e465d6	169	return _error->Error(_("Method %s did not start correctly"),Method.c_str());
3b5421b4 AL	170
3b5421b4 AL	171	RunMessages();
8b89e57f AL	172	if (OwnerQ != 0)
8b89e57f AL	173	SendConfiguration();
3b5421b4 AL	174
	175	return true;
	176	}
	177	/}}}/
	178	// Worker::ReadMessages - Read all pending messages into the list /{{{/
	179	// ---------------------------------------------------------------------
0a8a80e5	180	/* */
3b5421b4 AL	181	bool pkgAcquire::Worker::ReadMessages()
3b5421b4 AL	182	{
0a8a80e5 AL	183	if (::ReadMessages(InFd,MessageQueue) == false)
0a8a80e5 AL	184	return MethodFailure();
3b5421b4 AL	185	return true;
	186	}
	187	/}}}/
3b5421b4 AL	188	// Worker::RunMessage - Empty the message queue /{{{/
	189	// ---------------------------------------------------------------------
	190	/* This takes the messages from the message queue and runs them through
	191	the parsers in order. */
	192	bool pkgAcquire::Worker::RunMessages()
	193	{
	194	while (MessageQueue.empty() == false)
	195	{
	196	string Message = MessageQueue.front();
	197	MessageQueue.erase(MessageQueue.begin());
0a8a80e5 AL	198
	199	if (Debug == true)
	200	clog << " <- " << Access << ':' << QuoteString(Message,"\n") << endl;
3b5421b4 AL	201
	202	// Fetch the message number
	203	char *End;
	204	int Number = strtol(Message.c_str(),&End,10);
	205	if (End == Message.c_str())
	206	return _error->Error("Invalid message from method %s: %s",Access.c_str(),Message.c_str());
	207
c88edf1d AL	208	string URI = LookupTag(Message,"URI");
	209	pkgAcquire::Queue::QItem *Itm = 0;
	210	if (URI.empty() == false)
	211	Itm = OwnerQ->FindItem(URI,this);
196fd136 MV	212
	213	// update used mirror
	214	string UsedMirror = LookupTag(Message,"UsedMirror", "");
	215	if (!UsedMirror.empty() &&
	216	Itm &&
	217	Itm->Description.find(" ") != string::npos)
	218	{
	219	Itm->Description.replace(0, Itm->Description.find(" "), UsedMirror);
	220	// FIXME: will we need this as well?
	221	//Itm->ShortDesc = UsedMirror;
	222	}
bfd22fc0	223
3b5421b4 AL	224	// Determine the message number and dispatch
	225	switch (Number)
	226	{
0a8a80e5	227	// 100 Capabilities
3b5421b4 AL	228	case 100:
	229	if (Capabilities(Message) == false)
	230	return _error->Error("Unable to process Capabilities message from %s",Access.c_str());
	231	break;
0a8a80e5 AL	232
	233	// 101 Log
	234	case 101:
	235	if (Debug == true)
	236	clog << " <- (log) " << LookupTag(Message,"Message") << endl;
	237	break;
	238
	239	// 102 Status
	240	case 102:
	241	Status = LookupTag(Message,"Message");
	242	break;
	243
15d7e515 MV	244	// 103 Redirect
	245	case 103:
	246	{
	247	if (Itm == 0)
	248	{
	249	_error->Error("Method gave invalid 103 Redirect message");
	250	break;
	251	}
	252
	253	string NewURI = LookupTag(Message,"New-URI",URI.c_str());
	254	Itm->URI = NewURI;
5674f6b3 RG	255
	256	ItemDone();
	257
	258	pkgAcquire::Item *Owner = Itm->Owner;
	259	pkgAcquire::ItemDesc Desc = *Itm;
	260
	261	// Change the status so that it can be dequeued
	262	Owner->Status = pkgAcquire::Item::StatIdle;
	263	// Mark the item as done (taking care of all queues)
	264	// and then put it in the main queue again
	265	OwnerQ->ItemDone(Itm);
	266	OwnerQ->Owner->Enqueue(Desc);
	267
	268	if (Log != 0)
	269	Log->Done(Desc);
15d7e515 MV	270	break;
	271	}
	272
0a8a80e5 AL	273	// 200 URI Start
0a8a80e5 AL	274	case 200:
c88edf1d AL	275	{
	276	if (Itm == 0)
	277	{
93bf083d	278	_error->Error("Method gave invalid 200 URI Start message");
c88edf1d AL	279	break;
c88edf1d AL	280	}
8267fe24	281
c88edf1d AL	282	CurrentItem = Itm;
c88edf1d AL	283	CurrentSize = 0;
650faab0 DK	284	TotalSize = strtoull(LookupTag(Message,"Size","0").c_str(), NULL, 10);
	285	ResumePoint = strtoull(LookupTag(Message,"Resume-Point","0").c_str(), NULL, 10);
	286	Itm->Owner->Start(Message,strtoull(LookupTag(Message,"Size","0").c_str(), NULL, 10));
8b75eb1c	287
c5ccf175 AL	288	// Display update before completion
	289	if (Log != 0 && Log->MorePulses == true)
	290	Log->Pulse(Itm->Owner->GetOwner());
	291
8267fe24 AL	292	if (Log != 0)
	293	Log->Fetch(*Itm);
	294
c88edf1d AL	295	break;
c88edf1d AL	296	}
0a8a80e5 AL	297
	298	// 201 URI Done
	299	case 201:
c88edf1d AL	300	{
	301	if (Itm == 0)
	302	{
93bf083d	303	_error->Error("Method gave invalid 201 URI Done message");
c88edf1d AL	304	break;
c88edf1d AL	305	}
c5ccf175	306
bfd22fc0	307	pkgAcquire::Item *Owner = Itm->Owner;
8267fe24	308	pkgAcquire::ItemDesc Desc = *Itm;
c5ccf175 AL	309
	310	// Display update before completion
	311	if (Log != 0 && Log->MorePulses == true)
	312	Log->Pulse(Owner->GetOwner());
	313
be4401bf	314	OwnerQ->ItemDone(Itm);
650faab0	315	unsigned long long const ServerSize = strtoull(LookupTag(Message,"Size","0").c_str(), NULL, 10);
1dea08eb MV	316	bool isHit = StringToBool(LookupTag(Message,"IMS-Hit"),false) \|\|
	317	StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false);
	318	// Using the https method the server might return 200, but the
	319	// If-Modified-Since condition is not satsified, libcurl will
	320	// discard the download. In this case, however, TotalSize will be
	321	// set to the actual size of the file, while ServerSize will be set
	322	// to 0. Therefore, if the item is marked as a hit and the
	323	// downloaded size (ServerSize) is 0, we ignore TotalSize.
	324	if (TotalSize != 0 && (!isHit \|\| ServerSize != 0) && ServerSize != TotalSize)
73da43e9	325	_error->Warning("Size of file %s is not what the server reported %s %llu",
378fcbd6	326	Owner->DestFile.c_str(), LookupTag(Message,"Size","0").c_str(),TotalSize);
bdae53f1	327
8a8feb29	328	// see if there is a hash to verify
b3501edb DK	329	HashStringList RecivedHashes;
	330	HashStringList expectedHashes = Owner->HashSums();
	331	for (HashStringList::const_iterator hs = expectedHashes.begin(); hs != expectedHashes.end(); ++hs)
8a8feb29	332	{
b3501edb DK	333	std::string const tagname = hs->HashType() + "-Hash";
	334	std::string const hashsum = LookupTag(Message, tagname.c_str());
	335	if (hashsum.empty() == false)
	336	RecivedHashes.push_back(HashString(hs->HashType(), hashsum));
	337	}
	338
	339	if(_config->FindB("Debug::pkgAcquire::Auth", false) == true)
	340	{
	341	std::clog << "201 URI Done: " << Owner->DescURI() << endl
	342	<< "RecivedHash:" << endl;
	343	for (HashStringList::const_iterator hs = RecivedHashes.begin(); hs != RecivedHashes.end(); ++hs)
	344	std::clog << "\t- " << hs->toStr() << std::endl;
	345	std::clog << "ExpectedHash:" << endl;
	346	for (HashStringList::const_iterator hs = expectedHashes.begin(); hs != expectedHashes.end(); ++hs)
	347	std::clog << "\t- " << hs->toStr() << std::endl;
	348	std::clog << endl;
8a8feb29	349	}
b3501edb	350	Owner->Done(Message, ServerSize, RecivedHashes, Config);
8267fe24	351	ItemDone();
b3501edb	352
8267fe24 AL	353	// Log that we are done
	354	if (Log != 0)
	355	{
1dea08eb	356	if (isHit)
c46824ce AL	357	{
	358	/* Hide 'hits' for local only sources - we also manage to
	359	hide gets */
	360	if (Config->LocalOnly == false)
	361	Log->IMSHit(Desc);
	362	}
8267fe24 AL	363	else
8267fe24 AL	364	Log->Done(Desc);
18ef0a78	365	}
c88edf1d AL	366	break;
c88edf1d AL	367	}
0a8a80e5 AL	368
	369	// 400 URI Failure
	370	case 400:
c88edf1d AL	371	{
	372	if (Itm == 0)
	373	{
93bf083d	374	_error->Error("Method gave invalid 400 URI Failure message");
c88edf1d AL	375	break;
	376	}
	377
c5ccf175 AL	378	// Display update before completion
	379	if (Log != 0 && Log->MorePulses == true)
	380	Log->Pulse(Itm->Owner->GetOwner());
	381
bfd22fc0	382	pkgAcquire::Item *Owner = Itm->Owner;
8267fe24	383	pkgAcquire::ItemDesc Desc = *Itm;
c88edf1d	384	OwnerQ->ItemDone(Itm);
7e5f33eb MV	385
	386	// set some status
	387	if(LookupTag(Message,"FailReason") == "Timeout" \|\|
	388	LookupTag(Message,"FailReason") == "TmpResolveFailure" \|\|
f0983ff2	389	LookupTag(Message,"FailReason") == "ResolveFailure" \|\|
7e5f33eb MV	390	LookupTag(Message,"FailReason") == "ConnectionRefused")
	391	Owner->Status = pkgAcquire::Item::StatTransientNetworkError;
	392
7d8afa39	393	Owner->Failed(Message,Config);
8267fe24	394	ItemDone();
7d8afa39	395
8267fe24 AL	396	if (Log != 0)
8267fe24 AL	397	Log->Fail(Desc);
7d8afa39	398
c88edf1d AL	399	break;
c88edf1d AL	400	}
0a8a80e5 AL	401
	402	// 401 General Failure
	403	case 401:
b2e465d6	404	_error->Error("Method %s General failure: %s",Access.c_str(),LookupTag(Message,"Message").c_str());
0a8a80e5	405	break;
542ec555 AL	406
	407	// 403 Media Change
	408	case 403:
	409	MediaChange(Message);
	410	break;
3b5421b4 AL	411	}
	412	}
	413	return true;
	414	}
	415	/}}}/
	416	// Worker::Capabilities - 100 Capabilities handler /{{{/
	417	// ---------------------------------------------------------------------
	418	/* This parses the capabilities message and dumps it into the configuration
	419	structure. */
	420	bool pkgAcquire::Worker::Capabilities(string Message)
	421	{
	422	if (Config == 0)
	423	return true;
	424
	425	Config->Version = LookupTag(Message,"Version");
	426	Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
0a8a80e5 AL	427	Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
0a8a80e5 AL	428	Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
e331f6ed	429	Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
8e5fc8f5	430	Config->NeedsCleanup = StringToBool(LookupTag(Message,"Needs-Cleanup"),false);
459681d3	431	Config->Removable = StringToBool(LookupTag(Message,"Removable"),false);
3b5421b4 AL	432
	433	// Some debug text
	434	if (Debug == true)
	435	{
	436	clog << "Configured access method " << Config->Access << endl;
459681d3 AL	437	clog << "Version:" << Config->Version <<
	438	" SingleInstance:" << Config->SingleInstance <<
	439	" Pipeline:" << Config->Pipeline <<
	440	" SendConfig:" << Config->SendConfig <<
	441	" LocalOnly: " << Config->LocalOnly <<
	442	" NeedsCleanup: " << Config->NeedsCleanup <<
	443	" Removable: " << Config->Removable << endl;
3b5421b4 AL	444	}
3b5421b4 AL	445
542ec555 AL	446	return true;
	447	}
	448	/}}}/
	449	// Worker::MediaChange - Request a media change /{{{/
	450	// ---------------------------------------------------------------------
	451	/* */
	452	bool pkgAcquire::Worker::MediaChange(string Message)
	453	{
80a26ed1 MV	454	int status_fd = _config->FindI("APT::Status-Fd",-1);
	455	if(status_fd > 0)
	456	{
	457	string Media = LookupTag(Message,"Media");
	458	string Drive = LookupTag(Message,"Drive");
	459	ostringstream msg,status;
1a82c63e MV	460	ioprintf(msg,_("Please insert the disc labeled: "
	461	"'%s' "
	462	"in the drive '%s' and press enter."),
	463	Media.c_str(),Drive.c_str());
80a26ed1	464	status << "media-change: " // message
1a82c63e MV	465	<< Media << ":" // media
	466	<< Drive << ":" // drive
	467	<< msg.str() // l10n message
80a26ed1	468	<< endl;
31bda500 DK	469
31bda500 DK	470	std::string const dlstatus = status.str();
d68d65ad	471	FileFd::Write(status_fd, dlstatus.c_str(), dlstatus.size());
80a26ed1 MV	472	}
80a26ed1 MV	473
542ec555 AL	474	if (Log == 0 \|\| Log->MediaChange(LookupTag(Message,"Media"),
	475	LookupTag(Message,"Drive")) == false)
	476	{
	477	char S[300];
96bc43c4	478	snprintf(S,sizeof(S),"603 Media Changed\nFailed: true\n\n");
542ec555 AL	479	if (Debug == true)
	480	clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
	481	OutQueue += S;
	482	OutReady = true;
	483	return true;
	484	}
	485
	486	char S[300];
96bc43c4	487	snprintf(S,sizeof(S),"603 Media Changed\n\n");
542ec555 AL	488	if (Debug == true)
	489	clog << " -> " << Access << ':' << QuoteString(S,"\n") << endl;
	490	OutQueue += S;
	491	OutReady = true;
3b5421b4 AL	492	return true;
3b5421b4 AL	493	}
0118833a	494	/}}}/
0a8a80e5 AL	495	// Worker::SendConfiguration - Send the config to the method /{{{/
	496	// ---------------------------------------------------------------------
	497	/* */
	498	bool pkgAcquire::Worker::SendConfiguration()
	499	{
	500	if (Config->SendConfig == false)
	501	return true;
	502
	503	if (OutFd == -1)
	504	return false;
0a8a80e5	505
d280d03a	506	/* Write out all of the configuration directives by walking the
0a8a80e5	507	configuration tree */
d280d03a DK	508	std::ostringstream Message;
	509	Message << "601 Configuration\n";
	510	_config->Dump(Message, NULL, "Config-Item: %F=%V\n", false);
	511	Message << '\n';
0a8a80e5 AL	512
0a8a80e5 AL	513	if (Debug == true)
d280d03a DK	514	clog << " -> " << Access << ':' << QuoteString(Message.str(),"\n") << endl;
	515	OutQueue += Message.str();
	516	OutReady = true;
	517
0a8a80e5 AL	518	return true;
	519	}
	520	/}}}/
	521	// Worker::QueueItem - Add an item to the outbound queue /{{{/
	522	// ---------------------------------------------------------------------
	523	/* Send a URI Acquire message to the method */
	524	bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
	525	{
	526	if (OutFd == -1)
	527	return false;
	528
	529	string Message = "600 URI Acquire\n";
	530	Message.reserve(300);
	531	Message += "URI: " + Item->URI;
	532	Message += "\nFilename: " + Item->Owner->DestFile;
d003a557 DK	533	HashStringList const hsl = Item->Owner->HashSums();
	534	for (HashStringList::const_iterator hs = hsl.begin(); hs != hsl.end(); ++hs)
	535	Message += "\nExpected-" + hs->HashType() + ": " + hs->HashValue();
0a8a80e5 AL	536	Message += Item->Owner->Custom600Headers();
	537	Message += "\n\n";
	538
	539	if (Debug == true)
	540	clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
	541	OutQueue += Message;
	542	OutReady = true;
	543
	544	return true;
	545	}
	546	/}}}/
	547	// Worker::OutFdRead - Out bound FD is ready /{{{/
	548	// ---------------------------------------------------------------------
	549	/* */
	550	bool pkgAcquire::Worker::OutFdReady()
	551	{
ee7af1bd YY	552	int Res;
	553	do
	554	{
	555	Res = write(OutFd,OutQueue.c_str(),OutQueue.length());
	556	}
	557	while (Res < 0 && errno == EINTR);
	558
	559	if (Res <= 0)
0a8a80e5	560	return MethodFailure();
ee7af1bd YY	561
ee7af1bd YY	562	OutQueue.erase(0,Res);
0a8a80e5 AL	563	if (OutQueue.empty() == true)
	564	OutReady = false;
	565
	566	return true;
	567	}
	568	/}}}/
	569	// Worker::InFdRead - In bound FD is ready /{{{/
	570	// ---------------------------------------------------------------------
	571	/* */
	572	bool pkgAcquire::Worker::InFdReady()
	573	{
	574	if (ReadMessages() == false)
	575	return false;
	576	RunMessages();
	577	return true;
	578	}
	579	/}}}/
	580	// Worker::MethodFailure - Called when the method fails /{{{/
	581	// ---------------------------------------------------------------------
1e3f4083	582	/* This is called when the method is believed to have failed, probably because
0a8a80e5 AL	583	read returned -1. */
	584	bool pkgAcquire::Worker::MethodFailure()
	585	{
76d97c26 AL	586	_error->Error("Method %s has died unexpectedly!",Access.c_str());
76d97c26 AL	587
ab7f4d7c MV	588	// do not reap the child here to show meaningfull error to the user
ab7f4d7c MV	589	ExecWait(Process,Access.c_str(),false);
0a8a80e5 AL	590	Process = -1;
	591	close(InFd);
	592	close(OutFd);
	593	InFd = -1;
	594	OutFd = -1;
	595	OutReady = false;
	596	InReady = false;
	597	OutQueue = string();
	598	MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
	599
	600	return false;
	601	}
	602	/}}}/
8267fe24 AL	603	// Worker::Pulse - Called periodically /{{{/
	604	// ---------------------------------------------------------------------
	605	/* */
	606	void pkgAcquire::Worker::Pulse()
	607	{
	608	if (CurrentItem == 0)
	609	return;
542ec555	610
8267fe24 AL	611	struct stat Buf;
	612	if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
	613	return;
	614	CurrentSize = Buf.st_size;
18ef0a78 AL	615
	616	// Hmm? Should not happen...
	617	if (CurrentSize > TotalSize && TotalSize != 0)
	618	TotalSize = CurrentSize;
8267fe24 AL	619	}
	620	/}}}/
	621	// Worker::ItemDone - Called when the current item is finished /{{{/
	622	// ---------------------------------------------------------------------
	623	/* */
	624	void pkgAcquire::Worker::ItemDone()
	625	{
	626	CurrentItem = 0;
	627	CurrentSize = 0;
	628	TotalSize = 0;
	629	Status = string();
	630	}
	631	/}}}/