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

// -*- mode: cpp; mode: fold -*-
// Description								/*{{{*/
// $Id: acquire-worker.cc,v 1.13 1998/11/29 01:24:18 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							/*{{{*/
#ifdef __GNUG__
#pragma implementation "apt-pkg/acquire-worker.h"
#endif
#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 <strutl.h>

#include <sys/stat.h>
#include <unistd.h>
#include <signal.h>
#include <wait.h>
									/*}}}*/

// 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;

   Construct();   
}
									/*}}}*/
// Worker::Worker - Constructor for method config startup		/*{{{*/
// ---------------------------------------------------------------------
/* */
pkgAcquire::Worker::Worker(MethodConfig *Cnf)
{
   OwnerQ = 0;
   Config = Cnf;
   Access = Cnf->Access;
   CurrentItem = 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)
   {
      kill(Process,SIGINT);
      if (waitpid(Process,0,0) != Process)
	 _error->Warning("I waited but nothing was there!");
   }   
}
									/*}}}*/
// 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)
      return _error->Error("The method driver %s could not be found.",Method.c_str());

   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[0],true);
   
   // Fork off the process
   Process = fork();
   if (Process < 0)
   {
      cerr << "FATAL -> Failed to fork." << endl;
      exit(100);
   }

   // Spawn the subprocess
   if (Process == 0)
   {
      // Setup the FDs
      dup2(Pipes[1],STDOUT_FILENO);
      dup2(Pipes[2],STDIN_FILENO);
      dup2(((filebuf *)clog.rdbuf())->fd(),STDERR_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);
      
      // 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;
	    
	 // 200 URI Start
	 case 200:
	 {
	    if (Itm == 0)
	    {
	       _error->Error("Method gave invalid 200 URI Start message");
	       break;
	    }
	    
	    CurrentItem = Itm;
	    CurrentSize = 0;
	    TotalSize = atoi(LookupTag(Message,"Size","0").c_str());
	    Itm->Owner->Start(Message,atoi(LookupTag(Message,"Size","0").c_str()));
	    
	    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;
	    OwnerQ->ItemDone(Itm);
	    Owner->Done(Message,atoi(LookupTag(Message,"Size","0").c_str()),
					  LookupTag(Message,"MD5-Hash"));
	    ItemDone();
	    
	    // Log that we are done
	    if (Log != 0)
	    {
	       if (StringToBool(LookupTag(Message,"IMS-Hit"),false) == true ||
		   StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false) == true)
		  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;
	    }

	    pkgAcquire::Item *Owner = Itm->Owner;
	    pkgAcquire::ItemDesc Desc = *Itm;
	    OwnerQ->ItemDone(Itm);
	    Owner->Failed(Message);
	    ItemDone();
	    
	    if (Log != 0)
	       Log->Fail(Desc);
	    
	    break;
	 }	 
	 
	 // 401 General Failure
	 case 401:
	 _error->Error("Method %s General failure: %s",LookupTag(Message,"Message").c_str());
	 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);

   // 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 << endl;
   }
   
   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;
   
   string Message = "601 Configuration\n";
   Message.reserve(2000);

   /* Write out all of the configuration directives by walking the 
      configuration tree */
   const Configuration::Item *Top = _config->Tree(0);
   for (; Top != 0;)
   {
      if (Top->Value.empty() == false)
      {
	 string Line = "Config-Item: " + Top->FullTag() + "=";
	 Line += QuoteString(Top->Value,"\n") + '\n';
	 Message += Line;
      }
      
      if (Top->Child != 0)
      {
	 Top = Top->Child;
	 continue;
      }
      
      while (Top != 0 && Top->Next == 0)
	 Top = Top->Parent;
      if (Top != 0)
	 Top = Top->Next;
   }   
   Message += '\n';

   if (Debug == true)
      clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
   OutQueue += Message;
   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;
   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 = write(OutFd,OutQueue.begin(),OutQueue.length());
   if (Res <= 0)
      return MethodFailure();

   // Hmm.. this should never happen.
   if (Res < 0)
      return true;
   
   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 belived to have failed, probably because
   read returned -1. */
bool pkgAcquire::Worker::MethodFailure()
{
   cerr << "Method " << Access << " has died unexpectedly!" << endl;
   if (waitpid(Process,0,0) != Process)
      _error->Warning("I waited but nothing was there!");
   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;
}
									/*}}}*/
// 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 /{{{/
a72ace20	3	// $Id: acquire-worker.cc,v 1.13 1998/11/29 01:24:18 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 /{{{/
	15	#ifdef __GNUG__
	16	#pragma implementation "apt-pkg/acquire-worker.h"
3b5421b4	17	#endif
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>
	23	#include <strutl.h>
	24
8267fe24	25	#include <sys/stat.h>
3b5421b4 AL	26	#include <unistd.h>
3b5421b4 AL	27	#include <signal.h>
93641593	28	#include <wait.h>
3b5421b4 AL	29	/}}}/
	30
	31	// Worker::Worker - Constructor for Queue startup /{{{/
	32	// ---------------------------------------------------------------------
	33	/* */
8267fe24 AL	34	pkgAcquire::Worker::Worker(Queue Q,MethodConfig Cnf,
8267fe24 AL	35	pkgAcquireStatus *Log) : Log(Log)
3b5421b4 AL	36	{
3b5421b4 AL	37	OwnerQ = Q;
0a8a80e5 AL	38	Config = Cnf;
	39	Access = Cnf->Access;
	40	CurrentItem = 0;
3b5421b4 AL	41
	42	Construct();
	43	}
	44	/}}}/
	45	// Worker::Worker - Constructor for method config startup /{{{/
	46	// ---------------------------------------------------------------------
	47	/* */
	48	pkgAcquire::Worker::Worker(MethodConfig *Cnf)
	49	{
	50	OwnerQ = 0;
	51	Config = Cnf;
	52	Access = Cnf->Access;
0a8a80e5 AL	53	CurrentItem = 0;
0a8a80e5 AL	54
3b5421b4 AL	55	Construct();
	56	}
	57	/}}}/
	58	// Worker::Construct - Constructor helper /{{{/
	59	// ---------------------------------------------------------------------
	60	/* */
	61	void pkgAcquire::Worker::Construct()
	62	{
0a8a80e5 AL	63	NextQueue = 0;
0a8a80e5 AL	64	NextAcquire = 0;
3b5421b4 AL	65	Process = -1;
	66	InFd = -1;
	67	OutFd = -1;
0a8a80e5 AL	68	OutReady = false;
0a8a80e5 AL	69	InReady = false;
3b5421b4 AL	70	Debug = _config->FindB("Debug::pkgAcquire::Worker",false);
	71	}
	72	/}}}/
	73	// Worker::~Worker - Destructor /{{{/
	74	// ---------------------------------------------------------------------
	75	/* */
	76	pkgAcquire::Worker::~Worker()
	77	{
	78	close(InFd);
	79	close(OutFd);
	80
	81	if (Process > 0)
0a8a80e5	82	{
3b5421b4	83	kill(Process,SIGINT);
0a8a80e5 AL	84	if (waitpid(Process,0,0) != Process)
	85	_error->Warning("I waited but nothing was there!");
	86	}
3b5421b4 AL	87	}
	88	/}}}/
	89	// Worker::Start - Start the worker process /{{{/
	90	// ---------------------------------------------------------------------
	91	/* This forks the method and inits the communication channel */
	92	bool pkgAcquire::Worker::Start()
	93	{
	94	// Get the method path
	95	string Method = _config->FindDir("Dir::Bin::Methods") + Access;
	96	if (FileExists(Method) == false)
	97	return _error->Error("The method driver %s could not be found.",Method.c_str());
	98
	99	if (Debug == true)
	100	clog << "Starting method '" << Method << '\'' << endl;
	101
	102	// Create the pipes
	103	int Pipes[4] = {-1,-1,-1,-1};
	104	if (pipe(Pipes) != 0 \|\| pipe(Pipes+2) != 0)
	105	{
	106	_error->Errno("pipe","Failed to create IPC pipe to subprocess");
	107	for (int I = 0; I != 4; I++)
	108	close(Pipes[I]);
	109	return false;
	110	}
8b89e57f AL	111	for (int I = 0; I != 4; I++)
	112	SetCloseExec(Pipes[0],true);
	113
3b5421b4 AL	114	// Fork off the process
	115	Process = fork();
	116	if (Process < 0)
	117	{
	118	cerr << "FATAL -> Failed to fork." << endl;
	119	exit(100);
	120	}
	121
	122	// Spawn the subprocess
	123	if (Process == 0)
	124	{
	125	// Setup the FDs
	126	dup2(Pipes[1],STDOUT_FILENO);
	127	dup2(Pipes[2],STDIN_FILENO);
	128	dup2(((filebuf *)clog.rdbuf())->fd(),STDERR_FILENO);
3b5421b4 AL	129	SetCloseExec(STDOUT_FILENO,false);
	130	SetCloseExec(STDIN_FILENO,false);
	131	SetCloseExec(STDERR_FILENO,false);
	132
	133	const char *Args[2];
	134	Args[0] = Method.c_str();
	135	Args[1] = 0;
	136	execv(Args[0],(char **)Args);
	137	cerr << "Failed to exec method " << Args[0] << endl;
	138	exit(100);
	139	}
	140
	141	// Fix up our FDs
	142	InFd = Pipes[0];
	143	OutFd = Pipes[3];
	144	SetNonBlock(Pipes[0],true);
	145	SetNonBlock(Pipes[3],true);
	146	close(Pipes[1]);
	147	close(Pipes[2]);
0a8a80e5 AL	148	OutReady = false;
0a8a80e5 AL	149	InReady = true;
3b5421b4 AL	150
	151	// Read the configuration data
	152	if (WaitFd(InFd) == false \|\|
	153	ReadMessages() == false)
	154	return _error->Error("Method %s did not start correctly",Method.c_str());
	155
	156	RunMessages();
8b89e57f AL	157	if (OwnerQ != 0)
8b89e57f AL	158	SendConfiguration();
3b5421b4 AL	159
	160	return true;
	161	}
	162	/}}}/
	163	// Worker::ReadMessages - Read all pending messages into the list /{{{/
	164	// ---------------------------------------------------------------------
0a8a80e5	165	/* */
3b5421b4 AL	166	bool pkgAcquire::Worker::ReadMessages()
3b5421b4 AL	167	{
0a8a80e5 AL	168	if (::ReadMessages(InFd,MessageQueue) == false)
0a8a80e5 AL	169	return MethodFailure();
3b5421b4 AL	170	return true;
	171	}
	172	/}}}/
3b5421b4 AL	173	// Worker::RunMessage - Empty the message queue /{{{/
	174	// ---------------------------------------------------------------------
	175	/* This takes the messages from the message queue and runs them through
	176	the parsers in order. */
	177	bool pkgAcquire::Worker::RunMessages()
	178	{
	179	while (MessageQueue.empty() == false)
	180	{
	181	string Message = MessageQueue.front();
	182	MessageQueue.erase(MessageQueue.begin());
0a8a80e5 AL	183
	184	if (Debug == true)
	185	clog << " <- " << Access << ':' << QuoteString(Message,"\n") << endl;
3b5421b4 AL	186
	187	// Fetch the message number
	188	char *End;
	189	int Number = strtol(Message.c_str(),&End,10);
	190	if (End == Message.c_str())
	191	return _error->Error("Invalid message from method %s: %s",Access.c_str(),Message.c_str());
	192
c88edf1d AL	193	string URI = LookupTag(Message,"URI");
	194	pkgAcquire::Queue::QItem *Itm = 0;
	195	if (URI.empty() == false)
	196	Itm = OwnerQ->FindItem(URI,this);
bfd22fc0	197
3b5421b4 AL	198	// Determine the message number and dispatch
	199	switch (Number)
	200	{
0a8a80e5	201	// 100 Capabilities
3b5421b4 AL	202	case 100:
	203	if (Capabilities(Message) == false)
	204	return _error->Error("Unable to process Capabilities message from %s",Access.c_str());
	205	break;
0a8a80e5 AL	206
	207	// 101 Log
	208	case 101:
	209	if (Debug == true)
	210	clog << " <- (log) " << LookupTag(Message,"Message") << endl;
	211	break;
	212
	213	// 102 Status
	214	case 102:
	215	Status = LookupTag(Message,"Message");
	216	break;
	217
	218	// 200 URI Start
	219	case 200:
c88edf1d AL	220	{
	221	if (Itm == 0)
	222	{
93bf083d	223	_error->Error("Method gave invalid 200 URI Start message");
c88edf1d AL	224	break;
c88edf1d AL	225	}
8267fe24	226
c88edf1d AL	227	CurrentItem = Itm;
	228	CurrentSize = 0;
	229	TotalSize = atoi(LookupTag(Message,"Size","0").c_str());
8267fe24	230	Itm->Owner->Start(Message,atoi(LookupTag(Message,"Size","0").c_str()));
c88edf1d	231
8267fe24 AL	232	if (Log != 0)
	233	Log->Fetch(*Itm);
	234
c88edf1d AL	235	break;
c88edf1d AL	236	}
0a8a80e5 AL	237
	238	// 201 URI Done
	239	case 201:
c88edf1d AL	240	{
	241	if (Itm == 0)
	242	{
93bf083d	243	_error->Error("Method gave invalid 201 URI Done message");
c88edf1d AL	244	break;
	245	}
	246
bfd22fc0	247	pkgAcquire::Item *Owner = Itm->Owner;
8267fe24	248	pkgAcquire::ItemDesc Desc = *Itm;
be4401bf	249	OwnerQ->ItemDone(Itm);
bfd22fc0	250	Owner->Done(Message,atoi(LookupTag(Message,"Size","0").c_str()),
c88edf1d	251	LookupTag(Message,"MD5-Hash"));
8267fe24 AL	252	ItemDone();
	253
	254	// Log that we are done
	255	if (Log != 0)
	256	{
	257	if (StringToBool(LookupTag(Message,"IMS-Hit"),false) == true \|\|
	258	StringToBool(LookupTag(Message,"Alt-IMS-Hit"),false) == true)
	259	Log->IMSHit(Desc);
	260	else
	261	Log->Done(Desc);
	262	}
c88edf1d AL	263	break;
c88edf1d AL	264	}
0a8a80e5 AL	265
	266	// 400 URI Failure
	267	case 400:
c88edf1d AL	268	{
	269	if (Itm == 0)
	270	{
93bf083d	271	_error->Error("Method gave invalid 400 URI Failure message");
c88edf1d AL	272	break;
	273	}
	274
bfd22fc0	275	pkgAcquire::Item *Owner = Itm->Owner;
8267fe24	276	pkgAcquire::ItemDesc Desc = *Itm;
c88edf1d	277	OwnerQ->ItemDone(Itm);
bfd22fc0	278	Owner->Failed(Message);
8267fe24 AL	279	ItemDone();
	280
	281	if (Log != 0)
	282	Log->Fail(Desc);
	283
c88edf1d AL	284	break;
c88edf1d AL	285	}
0a8a80e5 AL	286
	287	// 401 General Failure
	288	case 401:
	289	_error->Error("Method %s General failure: %s",LookupTag(Message,"Message").c_str());
	290	break;
3b5421b4 AL	291	}
	292	}
	293	return true;
	294	}
	295	/}}}/
	296	// Worker::Capabilities - 100 Capabilities handler /{{{/
	297	// ---------------------------------------------------------------------
	298	/* This parses the capabilities message and dumps it into the configuration
	299	structure. */
	300	bool pkgAcquire::Worker::Capabilities(string Message)
	301	{
	302	if (Config == 0)
	303	return true;
	304
	305	Config->Version = LookupTag(Message,"Version");
	306	Config->SingleInstance = StringToBool(LookupTag(Message,"Single-Instance"),false);
0a8a80e5 AL	307	Config->Pipeline = StringToBool(LookupTag(Message,"Pipeline"),false);
0a8a80e5 AL	308	Config->SendConfig = StringToBool(LookupTag(Message,"Send-Config"),false);
e331f6ed	309	Config->LocalOnly = StringToBool(LookupTag(Message,"Local-Only"),false);
3b5421b4 AL	310
	311	// Some debug text
	312	if (Debug == true)
	313	{
	314	clog << "Configured access method " << Config->Access << endl;
0a8a80e5	315	clog << "Version:" << Config->Version << " SingleInstance:" <<
a72ace20	316	Config->SingleInstance <<
0a8a80e5 AL	317	" Pipeline:" << Config->Pipeline << " SendConfig:" <<
0a8a80e5 AL	318	Config->SendConfig << endl;
3b5421b4 AL	319	}
	320
	321	return true;
	322	}
0118833a	323	/}}}/
0a8a80e5 AL	324	// Worker::SendConfiguration - Send the config to the method /{{{/
	325	// ---------------------------------------------------------------------
	326	/* */
	327	bool pkgAcquire::Worker::SendConfiguration()
	328	{
	329	if (Config->SendConfig == false)
	330	return true;
	331
	332	if (OutFd == -1)
	333	return false;
	334
	335	string Message = "601 Configuration\n";
	336	Message.reserve(2000);
	337
	338	/* Write out all of the configuration directives by walking the
	339	configuration tree */
	340	const Configuration::Item *Top = _config->Tree(0);
	341	for (; Top != 0;)
	342	{
	343	if (Top->Value.empty() == false)
	344	{
	345	string Line = "Config-Item: " + Top->FullTag() + "=";
	346	Line += QuoteString(Top->Value,"\n") + '\n';
	347	Message += Line;
	348	}
	349
	350	if (Top->Child != 0)
	351	{
	352	Top = Top->Child;
	353	continue;
	354	}
	355
	356	while (Top != 0 && Top->Next == 0)
	357	Top = Top->Parent;
	358	if (Top != 0)
	359	Top = Top->Next;
	360	}
	361	Message += '\n';
	362
	363	if (Debug == true)
	364	clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
	365	OutQueue += Message;
	366	OutReady = true;
	367
	368	return true;
	369	}
	370	/}}}/
	371	// Worker::QueueItem - Add an item to the outbound queue /{{{/
	372	// ---------------------------------------------------------------------
	373	/* Send a URI Acquire message to the method */
	374	bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem *Item)
	375	{
	376	if (OutFd == -1)
	377	return false;
	378
	379	string Message = "600 URI Acquire\n";
	380	Message.reserve(300);
	381	Message += "URI: " + Item->URI;
	382	Message += "\nFilename: " + Item->Owner->DestFile;
	383	Message += Item->Owner->Custom600Headers();
	384	Message += "\n\n";
	385
	386	if (Debug == true)
	387	clog << " -> " << Access << ':' << QuoteString(Message,"\n") << endl;
388	OutQueue += Message;
389	OutReady = true;
390
391	return true;
392	}
393	/}}}/
394	// Worker::OutFdRead - Out bound FD is ready /{{{/
395	// ---------------------------------------------------------------------
396	/* */
397	bool pkgAcquire::Worker::OutFdReady()
398	{
399	int Res = write(OutFd,OutQueue.begin(),OutQueue.length());
400	if (Res <= 0)
401	return MethodFailure();
402
403	// Hmm.. this should never happen.
404	if (Res < 0)
405	return true;
406
407	OutQueue.erase(0,Res);
408	if (OutQueue.empty() == true)
409	OutReady = false;
410
411	return true;
412	}
413	/}}}/
414	// Worker::InFdRead - In bound FD is ready /{{{/
415	// ---------------------------------------------------------------------
416	/* */
417	bool pkgAcquire::Worker::InFdReady()
418	{
419	if (ReadMessages() == false)
420	return false;
421	RunMessages();
422	return true;
423	}
424	/}}}/
425	// Worker::MethodFailure - Called when the method fails /{{{/
426	// ---------------------------------------------------------------------
427	/* This is called when the method is belived to have failed, probably because
428	read returned -1. */
429	bool pkgAcquire::Worker::MethodFailure()
430	{
431	cerr << "Method " << Access << " has died unexpectedly!" << endl;
432	if (waitpid(Process,0,0) != Process)
433	_error->Warning("I waited but nothing was there!");
434	Process = -1;
435	close(InFd);
436	close(OutFd);
437	InFd = -1;
438	OutFd = -1;
439	OutReady = false;
440	InReady = false;
441	OutQueue = string();
442	MessageQueue.erase(MessageQueue.begin(),MessageQueue.end());
443
444	return false;
445	}
446	/}}}/
8267fe24 AL	447	// Worker::Pulse - Called periodically /{{{/
	448	// ---------------------------------------------------------------------
	449	/* */
	450	void pkgAcquire::Worker::Pulse()
	451	{
	452	if (CurrentItem == 0)
	453	return;
	454
	455
	456	struct stat Buf;
	457	if (stat(CurrentItem->Owner->DestFile.c_str(),&Buf) != 0)
	458	return;
	459	CurrentSize = Buf.st_size;
	460	}
	461	/}}}/
	462	// Worker::ItemDone - Called when the current item is finished /{{{/
	463	// ---------------------------------------------------------------------
	464	/* */
	465	void pkgAcquire::Worker::ItemDone()
	466	{
	467	CurrentItem = 0;
	468	CurrentSize = 0;
	469	TotalSize = 0;
	470	Status = string();
	471	}
	472	/}}}/