/*
* lftp - file transfer program
*
* Copyright (c) 1996-2013 by Alexander V. Lukyanov (lav@yars.free.net)
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
/* Usage notes:
*
* Call PreFilter() to add a filter to the beginning of the chain; these
* filters are initialized only once for all data. For example,
* PreFilter("wc -l")
*
*/
/*
* Implementation notes:
*
* Background things we can't get around:
* We must buffer (via FileCopy) output to a filter, since it might block.
*
* We must buffer the output from the filter to an output FileCopyPeer (ie.
* a URL), for the same reason.
*
* So, we're stuck with having up to two FileCopy's. (One to send, one to filter.)
*
* In some cases, we only need one: if the output is an FD, the filter can
* hook up directly and we can forget about that stage.
*
* In the case where we're outputting to a URL, we set up a FileCopy from a
* pipe to the URL, and then pretend we're just outputting to an FD (the
* pipe.) This means in the simple case of having no filters at all, writing
* to a URL or file, we send the data an extra time through a FileCopy and a
* pipe. That's a bit inefficient, but that's "cat file1 > file2"; that's
* normally done with "get file1 -o file2", so this shouldn't happen often.
*
* It's very important that if the output is stdout, any filters point directly
* at it, not through an extra copy: a pager, for example, will expect the output
* to be a TTY.
*
*/
#include <config.h>
#include "OutputJob.h"
#include "ArgV.h"
#include "FileCopy.h"
#include "CopyJob.h"
#include "url.h"
#include "misc.h"
#include "StatusLine.h"
#include "DummyProto.h"
#include <assert.h>
#include <unistd.h>
#include <errno.h>
#include <sys/ioctl.h>
#include <fcntl.h>
#include <stddef.h>
#define super Job
void OutputJob::InitCopy()
{
if(error)
return;
if(initialized)
return;
if(fa)
{
/* Set up a pipe sending data at the peer, so we can act like the FDStream
* constructor. */
int filter_pipe[2];
if(pipe(filter_pipe) == -1) {
// retry later
current->TimeoutS(1);
return;
}
FileCopyPeerFA *dst_peer = FileCopyPeerFA::New(fa.borrow(), fa_path, FA::STORE);
/* Status only for remote outputs. */
if(!strcmp(dst_peer->GetProto(), "file"))
no_status=true;
fcntl(filter_pipe[0],F_SETFL,O_NONBLOCK);
fcntl(filter_pipe[1],F_SETFL,O_NONBLOCK);
/* The output of the pipe (0) goes to the output FileCopy. */
FDStream *pipe_output = new FDStream(filter_pipe[0],"<filter-out>");
FileCopy *output_fc=FileCopy::New(new FileCopyPeerFDStream(pipe_output, FileCopyPeer::GET), dst_peer,false);
output=new CopyJob(output_fc, fa_path, a0);
output->NoStatus(no_status);
output_fd=new FDStream(filter_pipe[1],"<filter-in>");
pipe_output->CloseWhenDone();
output_fd->CloseWhenDone();
fa_path.set(0);
}
initialized=true;
if(Error())
return;
/* Clear the statusline, since we might change the pgrp if we create filters. */
ClearStatus();
/* Some legitimate uses produce broken pipe condition (cat|head).
* We still want to produce broken pipe if we're not piping, eg
* cat > pipe. */
if(IsFiltered())
fail_if_broken=false;
if(filter)
{
/* Create the global filter: */
OutputFilter *global = new OutputFilter(filter, output_fd.borrow());
output_fd=global;
}
/* Use a FileCopy to buffer our output to the filter: */
FileCopyPeerFDStream *out = new FileCopyPeerFDStream(output_fd.borrow(), FileCopyPeer::PUT);
FileCopy *input_fc = FileCopy::New(new FileCopyPeer(FileCopyPeer::GET), out, false);
if(!fail_if_broken)
input_fc->DontFailIfBroken();
input=new CopyJob(input_fc, xstring::format(_("%s (filter)"),a0.get()), filter?filter:a0);
if(!output)
output=input;
input->SetParentFg(this);
InputPeer()->SetDate(NO_DATE);
InputPeer()->SetSize(NO_SIZE);
input->GetCopy()->DontCopyDate();
input->NoStatus();
if(input != output)
{
output->SetParentFg(this);
OutputPeer()->SetDate(NO_DATE);
OutputPeer()->SetSize(NO_SIZE);
output->GetCopy()->DontCopyDate();
output->NoStatus();
}
if(is_stdout)
{
output->ClearStatusOnWrite();
output->GetCopy()->LineBuffered();
}
Timeout(0);
}
void OutputJob::Init(const char *_a0)
{
input=output=0;
initialized=false;
error=false;
no_status=false;
a0.set(_a0);
is_stdout=false;
fail_if_broken=true;
is_a_tty=false;
width=-1;
statusbar_redisplay=true;
}
/* Local (fd) output. */
OutputJob::OutputJob(FDStream *output_, const char *a0)
: output_fd(output_ ? output_ : new FDStream(1,"<stdout>"))
{
Init(a0);
if(output_)
{
/* We don't want to produce broken pipe when we're actually
* piping, since some legitimate uses produce broken pipe, eg
* cat|head. However, that's actually handled in InitCopy().
* User pipes aren't handled by us yet: instead of being set with
* SetFilter, they're being set up ahead of time and passed to
* us as an FDStream, so we don't really know if we're being filtered.
*
* So, until we handle pipes directly, disable broken pipe whenever
* we're being sent anywhere but stdout. */
fail_if_broken=false;
}
is_stdout=output_fd->usesfd(1);
is_a_tty=isatty(output_fd->fd);
width=fd_width(output_fd->fd);
/* We don't output status when outputting locally. */
no_status=true;
/* Make sure that if the output is going to fail, it fails early, so
* the parent doesn't start anything expensive (like begin downloading
* a file.) */
if(output_fd->getfd()==-1 && output_fd->error())
{
eprintf("%s: %s\n", a0, output_fd->error_text.get());
error=true;
}
}
OutputJob::OutputJob(const char *path, const char *a0, FileAccess *fa0)
: fa(fa0 ? fa0->Clone() : FileAccess::New("file")), fa_path(path)
{
Init(a0);
}
void OutputJob::PrepareToDie()
{
Bg();
AcceptSig(SIGTERM);
Delete(input);
if(input != output)
Delete(output);
input=0;
output=0;
super::PrepareToDie();
}
/* This is called to ask us "permission" to display a status line. */
bool OutputJob::ShowStatusLine(const SMTaskRef<StatusLine>& s)
{
/* If our output file is gone, or isn't stdout, we don't care. */
if(!output || !is_stdout)
return true;
/* If we're filtered, we never display at all. (We don't know anything about
* the filter's output; the only time we do is when we're outputting to a URL
* or a file, and that doesn't apply here.) */
if(IsFiltered())
return false;
/* If we're not line buffered, display only if the output CopyJob says to. */
if(!output->GetCopy()->IsLineBuffered())
return output->HasStatus();
/* We're line buffered, so we can output a status line without stomping
* on a partially output line.
*
* If we've output something recently, only send the output to the title,
* to avoid flickering status for no reason.
*/
if(!update_timer.Stopped()) {
s->NextUpdateTitleOnly();
return true;
}
/* If we're not reenabling the status bar, and the statusbar has
* been turned off (due to output being reenabled), only send to
* the title. */
if(!statusbar_redisplay && output->GetCopy()->WriteAllowed())
{
s->NextUpdateTitleOnly();
return true;
}
/* Don't disable write if there are data to be written in buffer */
if(output->GetCopy()->WriteAllowed() && output->GetCopy()->WritePending())
return false;
/* There hasn't been output in a while. Stop the output again,
* so the FileCopy will clear the StatusLine when there's more data. */
output->GetCopy()->AllowWrite(false);
return true;
}
/* Get our contribution to the status line, which is just the output
* status, if any. Input status is the job of the user object. */
const char *OutputJob::Status(const StatusLine *s)
{
if(no_status)
return "";
/* Never show anything if we havn't even received any data yet; it won't
* start connecting until then, so it's not interesting. */
if(!initialized)
return "";
/* Use the status from the output CopyJob. It'll be the one that's connecting
* to a host, if applicable. */
return output->Status(s,true);
}
void OutputJob::PutEOF()
{
if(Error())
return;
/* Make sure we've sent at least one (empty) block. This ensures
* that we always start the input->output code path. */
Put("", 0);
/* Send an EOF to the input peer; it'll send an EOF to the output peer
* when all of its data is actually sent. */
if(InputPeer())
InputPeer()->PutEOF();
else if(tmp_buf)
tmp_buf->PutEOF();
else
abort();
}
/* add a filter to the beginning of the list */
void OutputJob::PreFilter(const char *newfilter)
{
if(!filter)
filter.set(newfilter);
else
{
char *old_filter=alloca_strdup(filter);
filter.vset(newfilter," | ",old_filter,NULL);
}
}
/* Return the width of the output. If there's a filter, we can either
* return -1 (we might be piping through "sed", changing the width),
* or the width we know (which is sane for most pagers.) I'm not sure
* which is better. */
int OutputJob::GetWidth() const
{
if(IsFiltered())
return -1;
return width;
}
/* Return true if the output is going directly to a TTY. */
bool OutputJob::IsTTY() const
{
if(IsFiltered())
return false;
return is_a_tty;
}
static const SMTaskRef<FileCopyPeer> null_peer;
/* Get the input FileCopyPeer; this is the buffer we write to. */
const SMTaskRef<FileCopyPeer>& OutputJob::InputPeer() const
{
if(input)
return input->GetGet();
return null_peer;
}
/* Get the output FileCopyPeer (the FileCopyPeer that's doing the final output). */
const SMTaskRef<FileCopyPeer>& OutputJob::OutputPeer() const
{
if(output)
return output->GetPut();
return null_peer;
}
/* We're done if the output is finished, or on error. */
int OutputJob::Done()
{
if(Error())
return true;
if(!initialized)
return false;
if(input && !input->Done())
return false;
if(output && !output->Done())
return false;
return true;
}
int OutputJob::Do()
{
if(!initialized && tmp_buf)
InitCopy();
return STALL;
}
/* Don't register errors until they're actually printed by
* the sub-job (ie. it's also Done()). */
bool OutputJob::Error()
{
if(error)
return true;
if(input && input->Error() && input->Done())
error=true;
if(output && input != output && output->Error() && output->Done())
error=true;
return error;
}
void OutputJob::Fg()
{
super::Fg();
if(input)
input->Fg();
if(output && input != output)
output->Fg();
}
void OutputJob::Bg()
{
if(output && input != output)
output->Bg();
if(input)
input->Bg();
super::Bg();
}
void OutputJob::SuspendInternal()
{
super::SuspendInternal();
if(input)
input->SuspendSlave();
if(output && input != output)
output->SuspendSlave();
}
void OutputJob::ResumeInternal()
{
if(input)
input->ResumeSlave();
if(output && input != output)
output->ResumeSlave();
super::ResumeInternal();
}
bool OutputJob::Full()
{
/* It'd be nicer to just check copy->GetGet()->IsSuspended(), since
* the FileCopy will suspend the Get end if the Put end gets filled.
* However, it won't do that until it actually tries to send something. */
int size = 0;
if(input)
{
if(input->GetPut())
size += input->GetPut()->Buffered();
if(input->GetGet())
size += input->GetGet()->Buffered();
if(input != output)
{
if(output->GetPut())
size += output->GetPut()->Buffered();
if(output->GetGet())
size += output->GetGet()->Buffered();
}
}
if(tmp_buf)
size += tmp_buf->Size();
return size >= 0x10000;
}
/* We'll actually go over the buffer limit here; that's OK; it's not a
* strict value. (It's not convenient to prevent that completely with
* Format(), either.) */
void OutputJob::Put(const char *buf,int size)
{
InitCopy();
if(Error())
return;
if(!InputPeer())
{
if(!tmp_buf)
tmp_buf=new Buffer;
tmp_buf->Put(buf,size);
return;
}
// InputPeer was inited, flush tmp_buf.
if(InputPeer() && tmp_buf)
{
Ref<Buffer> saved_buf(tmp_buf.borrow());
const char *b=0;
int s=0;
saved_buf->Get(&b,&s);
if(b && s>0)
Put(b,s);
if(saved_buf->Eof())
PutEOF();
}
update_timer.SetResource("cmd:status-interval",0);
off_t oldpos = InputPeer()->GetPos();
InputPeer()->Put(buf, size);
InputPeer()->SetPos(oldpos);
}
void OutputJob::Format(const char *f,...)
{
va_list v;
va_start(v,f);
const xstring& str=xstring::vformat(f,v);
va_end(v);
Put(str);
}
/* Propagate signals down to our child processes. */
int OutputJob::AcceptSig(int sig)
{
int m=MOVED;
if(sig == SIGTERM || sig == SIGINT)
m=WANTDIE;
/* If we have an input copier right now, it'll contain the top filter
* (which is linked to all other filters), so send it the signal. */
if(input)
input->AcceptSig(sig);
/* Otherwise, the only filters we have running are in output_fd. */
else if(output_fd)
output_fd->Kill(sig);
if(sig!=SIGCONT)
AcceptSig(SIGCONT);
return m;
}