google code style, and when paused, don't die after 10 seconds

This commit is contained in:
Isaac Connor 2018-04-12 13:14:00 -04:00
parent 2af539e916
commit b974b4dcd1
3 changed files with 206 additions and 235 deletions

View File

@ -43,30 +43,30 @@
bool EventStream::loadInitialEventData( int monitor_id, time_t event_time ) {
static char sql[ZM_SQL_SML_BUFSIZ];
snprintf( sql, sizeof(sql), "select Id from Events where MonitorId = %d and unix_timestamp( EndTime ) > %ld order by Id asc limit 1", monitor_id, event_time );
snprintf(sql, sizeof(sql), "SELECT Id FROM Events WHERE MonitorId = %d AND unix_timestamp(EndTime) > %ld ORDER BY Id ASC LIMIT 1", monitor_id, event_time);
if ( mysql_query( &dbconn, sql ) ) {
Error( "Can't run query: %s", mysql_error( &dbconn ) );
exit( mysql_errno( &dbconn ) );
if ( mysql_query(&dbconn, sql) ) {
Error("Can't run query: %s", mysql_error(&dbconn));
exit(mysql_errno(&dbconn));
}
MYSQL_RES *result = mysql_store_result( &dbconn );
MYSQL_RES *result = mysql_store_result(&dbconn);
if ( !result ) {
Error( "Can't use query result: %s", mysql_error( &dbconn ) );
Error("Can't use query result: %s", mysql_error(&dbconn));
exit( mysql_errno( &dbconn ) );
}
MYSQL_ROW dbrow = mysql_fetch_row( result );
MYSQL_ROW dbrow = mysql_fetch_row(result);
if ( mysql_errno( &dbconn ) ) {
Error( "Can't fetch row: %s", mysql_error( &dbconn ) );
exit( mysql_errno( &dbconn ) );
if ( mysql_errno(&dbconn) ) {
Error("Can't fetch row: %s", mysql_error(&dbconn));
exit( mysql_errno(&dbconn));
}
int init_event_id = atoi( dbrow[0] );
int init_event_id = atoi(dbrow[0]);
mysql_free_result( result );
mysql_free_result(result);
loadEventData( init_event_id );
loadEventData(init_event_id);
if ( event_time ) {
curr_stream_time = event_time;
@ -778,20 +778,19 @@ void EventStream::runStream() {
checkInitialised();
if ( type == STREAM_JPEG )
fprintf( stdout, "Content-Type: multipart/x-mixed-replace;boundary=ZoneMinderFrame\r\n\r\n" );
fputs("Content-Type: multipart/x-mixed-replace;boundary=ZoneMinderFrame\r\n\r\n", stdout);
if ( !event_data ) {
sendTextFrame( "No event data found" );
exit( 0 );
sendTextFrame("No event data found");
exit(0);
}
Debug(3, "frame rate is: (%f)", (double)event_data->frame_count/event_data->duration );
updateFrameRate( (double)event_data->frame_count/event_data->duration );
Debug(3, "frame rate is: (%f)", (double)event_data->frame_count/event_data->duration);
updateFrameRate((double)event_data->frame_count/event_data->duration);
while( !zm_terminate ) {
gettimeofday( &now, NULL );
gettimeofday(&now, NULL);
unsigned int delta_us = 0;
send_frame = false;
@ -824,11 +823,11 @@ void EventStream::runStream() {
in_event = false;
}
if ( !in_event ) {
double actual_delta_time = TV_2_FLOAT( now ) - last_frame_sent;
double actual_delta_time = TV_2_FLOAT(now) - last_frame_sent;
if ( actual_delta_time > 1 ) {
static char frame_text[64];
snprintf( frame_text, sizeof(frame_text), "Time to next event = %d seconds", (int)time_to_event );
if ( !sendTextFrame( frame_text ) )
snprintf(frame_text, sizeof(frame_text), "Time to next event = %d seconds", (int)time_to_event);
if ( !sendTextFrame(frame_text) )
zm_terminate = true;
}
//else
@ -857,16 +856,16 @@ void EventStream::runStream() {
send_frame = true;
} else if ( !send_frame ) {
// We are paused, and doing nothing
double actual_delta_time = TV_2_FLOAT( now ) - last_frame_sent;
double actual_delta_time = TV_2_FLOAT(now) - last_frame_sent;
if ( actual_delta_time > MAX_STREAM_DELAY ) {
// Send keepalive
Debug( 2, "Sending keepalive frame" );
Debug(2, "Sending keepalive frame");
send_frame = true;
}
}
if ( send_frame )
if ( !sendFrame( delta_us ) )
if ( !sendFrame(delta_us) )
zm_terminate = true;
curr_stream_time = frame_data->timestamp;

View File

@ -27,28 +27,28 @@
#include <arpa/inet.h>
#include <glob.h>
bool MonitorStream::checkSwapPath( const char *path, bool create_path ) {
bool MonitorStream::checkSwapPath(const char *path, bool create_path) {
struct stat stat_buf;
if ( stat( path, &stat_buf ) < 0 ) {
if ( stat(path, &stat_buf) < 0 ) {
if ( create_path && errno == ENOENT ) {
Debug( 3, "Swap path '%s' missing, creating", path );
if ( mkdir( path, 0755 ) ) {
Error( "Can't mkdir %s: %s", path, strerror(errno));
return( false );
Debug(3, "Swap path '%s' missing, creating", path);
if ( mkdir(path, 0755) ) {
Error("Can't mkdir %s: %s", path, strerror(errno));
return false;
}
if ( stat( path, &stat_buf ) < 0 ) {
Error( "Can't stat '%s': %s", path, strerror(errno) );
return( false );
if ( stat(path, &stat_buf) < 0 ) {
Error("Can't stat '%s': %s", path, strerror(errno));
return false;
}
} else {
Error( "Can't stat '%s': %s", path, strerror(errno) );
return( false );
Error("Can't stat '%s': %s", path, strerror(errno));
return false;
}
}
if ( !S_ISDIR(stat_buf.st_mode) ) {
Error( "Swap image path '%s' is not a directory", path );
return( false );
Error("Swap image path '%s' is not a directory", path);
return false;
}
uid_t uid = getuid();
@ -67,31 +67,26 @@ bool MonitorStream::checkSwapPath( const char *path, bool create_path ) {
}
if ( (stat_buf.st_mode & mask) != mask ) {
Error( "Insufficient permissions on swap image path '%s'", path );
return( false );
Error("Insufficient permissions on swap image path '%s'", path);
return false;
}
return( true );
return true;
} // end bool MonitorStream::checkSwapPath( const char *path, bool create_path )
void MonitorStream::processCommand( const CmdMsg *msg ) {
void MonitorStream::processCommand(const CmdMsg *msg) {
Debug( 2, "Got message, type %d, msg %d", msg->msg_type, msg->msg_data[0] );
// Check for incoming command
switch( (MsgCommand)msg->msg_data[0] ) {
case CMD_PAUSE :
Debug(1, "Got PAUSE command");
// Set paused flag
paused = true;
// Set delayed flag
delayed = true;
last_frame_sent = TV_2_FLOAT(now);
break;
case CMD_PLAY :
Debug(1, "Got PLAY command");
if ( paused ) {
// Clear paused flag
paused = false;
// Set delayed_play flag
delayed = true;
}
replay_rate = ZM_RATE_BASE;
@ -99,27 +94,20 @@ void MonitorStream::processCommand( const CmdMsg *msg ) {
case CMD_VARPLAY :
Debug(1, "Got VARPLAY command");
if ( paused ) {
// Clear paused flag
paused = false;
// Set delayed_play flag
delayed = true;
}
replay_rate = ntohs(((unsigned char)msg->msg_data[2]<<8)|(unsigned char)msg->msg_data[1])-32768;
break;
case CMD_STOP :
Debug(1, "Got STOP command");
// Clear paused flag
paused = false;
// Clear delayed_play flag
delayed = false;
break;
case CMD_FASTFWD :
Debug(1, "Got FAST FWD command");
if ( paused ) {
// Clear paused flag
paused = false;
// Set delayed_play flag
delayed = true;
}
// Set play rate
@ -144,32 +132,22 @@ void MonitorStream::processCommand( const CmdMsg *msg ) {
break;
case CMD_SLOWFWD :
Debug( 1, "Got SLOW FWD command" );
// Set paused flag
paused = true;
// Set delayed flag
delayed = true;
// Set play rate
replay_rate = ZM_RATE_BASE;
// Set step
step = 1;
break;
case CMD_SLOWREV :
Debug( 1, "Got SLOW REV command" );
// Set paused flag
paused = true;
// Set delayed flag
delayed = true;
// Set play rate
replay_rate = ZM_RATE_BASE;
// Set step
step = -1;
break;
case CMD_FASTREV :
Debug( 1, "Got FAST REV command" );
if ( paused ) {
// Clear paused flag
paused = false;
// Set delayed_play flag
delayed = true;
}
// Set play rate
@ -239,20 +217,20 @@ void MonitorStream::processCommand( const CmdMsg *msg ) {
case CMD_PAN :
x = ((unsigned char)msg->msg_data[1]<<8)|(unsigned char)msg->msg_data[2];
y = ((unsigned char)msg->msg_data[3]<<8)|(unsigned char)msg->msg_data[4];
Debug( 1, "Got PAN command, to %d,%d", x, y );
Debug(1, "Got PAN command, to %d,%d", x, y);
break;
case CMD_SCALE :
scale = ((unsigned char)msg->msg_data[1]<<8)|(unsigned char)msg->msg_data[2];
Debug( 1, "Got SCALE command, to %d", scale );
Debug(1, "Got SCALE command, to %d", scale);
break;
case CMD_QUIT :
Info ("User initiated exit - CMD_QUIT");
Info("User initiated exit - CMD_QUIT");
break;
case CMD_QUERY :
Debug( 1, "Got QUERY command, sending STATUS" );
Debug(1, "Got QUERY command, sending STATUS");
break;
default :
Error( "Got unexpected command %d", msg->msg_data[0] );
Error("Got unexpected command %d", msg->msg_data[0]);
break;
} // end switch command
@ -285,7 +263,7 @@ void MonitorStream::processCommand( const CmdMsg *msg ) {
//status_data.enabled = monitor->shared_data->active;
status_data.enabled = monitor->trigger_data->trigger_state!=Monitor::TRIGGER_OFF;
status_data.forced = monitor->trigger_data->trigger_state==Monitor::TRIGGER_ON;
Debug( 2, "Buffer Level:%d, Delayed:%d, Paused:%d, Rate:%d, delay:%.3f, Zoom:%d, Enabled:%d Forced:%d",
Debug(2, "Buffer Level:%d, Delayed:%d, Paused:%d, Rate:%d, delay:%.3f, Zoom:%d, Enabled:%d Forced:%d",
status_data.buffer_level,
status_data.delayed,
status_data.paused,
@ -298,7 +276,7 @@ void MonitorStream::processCommand( const CmdMsg *msg ) {
DataMsg status_msg;
status_msg.msg_type = MSG_DATA_WATCH;
memcpy( &status_msg.msg_data, &status_data, sizeof(status_data) );
memcpy(&status_msg.msg_data, &status_data, sizeof(status_data));
int nbytes = 0;
if ( (nbytes = sendto(sd, &status_msg, sizeof(status_msg), MSG_DONTWAIT, (sockaddr *)&rem_addr, sizeof(rem_addr))) < 0 ) {
//if ( errno != EAGAIN )
@ -307,7 +285,7 @@ void MonitorStream::processCommand( const CmdMsg *msg ) {
//exit( -1 );
}
}
Debug(2, "NUmber of bytes sent to (%s): (%d)", rem_addr.sun_path, nbytes );
Debug(2, "Number of bytes sent to (%s): (%d)", rem_addr.sun_path, nbytes);
// quit after sending a status, if this was a quit request
if ( (MsgCommand)msg->msg_data[0]==CMD_QUIT ) {
@ -316,10 +294,10 @@ Debug(2, "NUmber of bytes sent to (%s): (%d)", rem_addr.sun_path, nbytes );
}
Debug(2,"Updating framerate");
updateFrameRate( monitor->GetFPS() );
} // end void MonitorStream::processCommand( const CmdMsg *msg )
updateFrameRate(monitor->GetFPS());
} // end void MonitorStream::processCommand(const CmdMsg *msg)
bool MonitorStream::sendFrame( const char *filepath, struct timeval *timestamp ) {
bool MonitorStream::sendFrame(const char *filepath, struct timeval *timestamp) {
bool send_raw = ((scale>=ZM_SCALE_BASE)&&(zoom==ZM_SCALE_BASE));
if ( type != STREAM_JPEG )
@ -328,71 +306,70 @@ bool MonitorStream::sendFrame( const char *filepath, struct timeval *timestamp )
send_raw = false;
if ( !send_raw ) {
Image temp_image( filepath );
Image temp_image(filepath);
return( sendFrame( &temp_image, timestamp ) );
return sendFrame(&temp_image, timestamp);
} else {
int img_buffer_size = 0;
static unsigned char img_buffer[ZM_MAX_IMAGE_SIZE];
FILE *fdj = NULL;
if ( (fdj = fopen( filepath, "r" )) ) {
img_buffer_size = fread( img_buffer, 1, sizeof(img_buffer), fdj );
fclose( fdj );
if ( (fdj = fopen(filepath, "r")) ) {
img_buffer_size = fread(img_buffer, 1, sizeof(img_buffer), fdj);
fclose(fdj);
} else {
Error( "Can't open %s: %s", filepath, strerror(errno) );
return( false );
Error("Can't open %s: %s", filepath, strerror(errno));
return false;
}
// Calculate how long it takes to actually send the frame
struct timeval frameStartTime;
gettimeofday( &frameStartTime, NULL );
gettimeofday(&frameStartTime, NULL);
fprintf( stdout, "--ZoneMinderFrame\r\n" );
fprintf( stdout, "Content-Length: %d\r\n", img_buffer_size );
fprintf( stdout, "Content-Type: image/jpeg\r\n\r\n" );
if ( fwrite( img_buffer, img_buffer_size, 1, stdout ) != 1 ) {
fputs("--ZoneMinderFrame\r\nContent-Type: image/jpeg\r\n\r\n", stdout );
fprintf(stdout, "Content-Length: %d\r\n", img_buffer_size);
if ( fwrite(img_buffer, img_buffer_size, 1, stdout) != 1 ) {
if ( ! zm_terminate )
Error( "Unable to send stream frame: %s", strerror(errno) );
return( false );
Error("Unable to send stream frame: %s", strerror(errno));
return false;
}
fprintf( stdout, "\r\n\r\n" );
fflush( stdout );
fputs("\r\n\r\n", stdout);
fflush(stdout);
struct timeval frameEndTime;
gettimeofday( &frameEndTime, NULL );
gettimeofday(&frameEndTime, NULL);
int frameSendTime = tvDiffMsec( frameStartTime, frameEndTime );
int frameSendTime = tvDiffMsec(frameStartTime, frameEndTime);
if ( frameSendTime > 1000/maxfps ) {
maxfps /= 2;
Info( "Frame send time %d msec too slow, throttling maxfps to %.2f", frameSendTime, maxfps );
Info("Frame send time %d msec too slow, throttling maxfps to %.2f", frameSendTime, maxfps);
}
last_frame_sent = TV_2_FLOAT( now );
last_frame_sent = TV_2_FLOAT(now);
return( true );
return true;
}
return( false );
}
return false;
} // end bool MonitorStream::sendFrame(const char *filepath, struct timeval *timestamp)
bool MonitorStream::sendFrame( Image *image, struct timeval *timestamp ) {
Image *send_image = prepareImage( image );
bool MonitorStream::sendFrame(Image *image, struct timeval *timestamp) {
Image *send_image = prepareImage(image);
if ( !config.timestamp_on_capture && timestamp )
monitor->TimestampImage( send_image, timestamp );
monitor->TimestampImage(send_image, timestamp);
#if HAVE_LIBAVCODEC
if ( type == STREAM_MPEG ) {
if ( !vid_stream ) {
vid_stream = new VideoStream( "pipe:", format, bitrate, effective_fps, send_image->Colours(), send_image->SubpixelOrder(), send_image->Width(), send_image->Height() );
fprintf( stdout, "Content-type: %s\r\n\r\n", vid_stream->MimeType() );
vid_stream = new VideoStream("pipe:", format, bitrate, effective_fps, send_image->Colours(), send_image->SubpixelOrder(), send_image->Width(), send_image->Height());
fprintf(stdout, "Content-type: %s\r\n\r\n", vid_stream->MimeType());
vid_stream->OpenStream();
}
static struct timeval base_time;
struct DeltaTimeval delta_time;
if ( !frame_count )
base_time = *timestamp;
DELTA_TIMEVAL( delta_time, *timestamp, base_time, DT_PREC_3 );
/* double pts = */ vid_stream->EncodeFrame( send_image->Buffer(), send_image->Size(), config.mpeg_timed_frames, delta_time.delta );
DELTA_TIMEVAL(delta_time, *timestamp, base_time, DT_PREC_3);
/* double pts = */ vid_stream->EncodeFrame(send_image->Buffer(), send_image->Size(), config.mpeg_timed_frames, delta_time.delta);
} else
#endif // HAVE_LIBAVCODEC
{
@ -403,36 +380,38 @@ bool MonitorStream::sendFrame( Image *image, struct timeval *timestamp ) {
// Calculate how long it takes to actually send the frame
struct timeval frameStartTime;
gettimeofday( &frameStartTime, NULL );
gettimeofday(&frameStartTime, NULL);
fprintf( stdout, "--ZoneMinderFrame\r\n" );
fputs("--ZoneMinderFrame\r\n", stdout);
switch( type ) {
case STREAM_JPEG :
send_image->EncodeJpeg( img_buffer, &img_buffer_size );
fprintf( stdout, "Content-Type: image/jpeg\r\n" );
send_image->EncodeJpeg(img_buffer, &img_buffer_size);
fputs("Content-Type: image/jpeg\r\n", stdout);
break;
case STREAM_RAW :
fprintf( stdout, "Content-Type: image/x-rgb\r\n" );
fputs("Content-Type: image/x-rgb\r\n", stdout);
img_buffer = (uint8_t*)send_image->Buffer();
img_buffer_size = send_image->Size();
break;
case STREAM_ZIP :
fprintf( stdout, "Content-Type: image/x-rgbz\r\n" );
fputs("Content-Type: image/x-rgbz\r\n",stdout);
unsigned long zip_buffer_size;
send_image->Zip( img_buffer, &zip_buffer_size );
send_image->Zip(img_buffer, &zip_buffer_size);
img_buffer_size = zip_buffer_size;
break;
default :
Fatal( "Unexpected frame type %d", type );
break;
Error("Unexpected frame type %d", type);
return false;
}
fprintf( stdout, "Content-Length: %d\r\n\r\n", img_buffer_size );
if ( fwrite( img_buffer, img_buffer_size, 1, stdout ) != 1 ) {
if ( !zm_terminate )
Error( "Unable to send stream frame: %s", strerror(errno) );
return( false );
fprintf(stdout, "Content-Length: %d\r\n\r\n", img_buffer_size);
if ( fwrite(img_buffer, img_buffer_size, 1, stdout) != 1 ) {
if ( !zm_terminate ){
// If the pipe was closed, we will get signalled SIGPIPE to exit, which will set zm_terminate
Error("Unable to send stream frame: %s", strerror(errno));
}
return false;
}
fprintf( stdout, "\r\n\r\n" );
fputs("\r\n\r\n",stdout);
fflush( stdout );
struct timeval frameEndTime;
@ -451,27 +430,27 @@ bool MonitorStream::sendFrame( Image *image, struct timeval *timestamp ) {
void MonitorStream::runStream() {
if ( type == STREAM_SINGLE ) {
// Not yet migrated over to stream class
SingleImage( scale );
SingleImage(scale);
return;
}
openComms();
if ( ! checkInitialised() ) {
if ( !checkInitialised() ) {
Error("Not initialized");
return;
}
updateFrameRate( monitor->GetFPS() );
updateFrameRate(monitor->GetFPS());
if ( type == STREAM_JPEG )
fprintf( stdout, "Content-Type: multipart/x-mixed-replace;boundary=ZoneMinderFrame\r\n\r\n" );
fputs("Content-Type: multipart/x-mixed-replace;boundary=ZoneMinderFrame\r\n\r\n", stdout);
// point to end which is theoretically not a valid value because all indexes are % image_buffer_count
unsigned int last_read_index = monitor->image_buffer_count;
time_t stream_start_time;
time( &stream_start_time );
time(&stream_start_time);
frame_count = 0;
@ -480,48 +459,43 @@ void MonitorStream::runStream() {
temp_read_index = temp_image_buffer_count;
temp_write_index = temp_image_buffer_count;
char *swap_path = 0;
std::string swap_path;
bool buffered_playback = false;
// 15 is the max length for the swap path suffix, /zmswap-whatever, assuming max 6 digits for monitor id
const int max_swap_len_suffix = 15;
int swap_path_length = staticConfig.PATH_SWAP.length() + 1; // +1 for NULL terminator
int subfolder1_length = snprintf(NULL, 0, "/zmswap-m%d", monitor->Id() ) + 1;
int subfolder2_length = snprintf(NULL, 0, "/zmswap-q%06d", connkey ) + 1;
int subfolder1_length = snprintf(NULL, 0, "/zmswap-m%d", monitor->Id()) + 1;
int subfolder2_length = snprintf(NULL, 0, "/zmswap-q%06d", connkey) + 1;
int total_swap_path_length = swap_path_length + subfolder1_length + subfolder2_length;
if ( connkey && playback_buffer > 0 ) {
if ( connkey && ( playback_buffer > 0 ) ) {
if ( total_swap_path_length + max_swap_len_suffix > PATH_MAX ) {
Error( "Swap Path is too long. %d > %d ", total_swap_path_length+max_swap_len_suffix, PATH_MAX );
Error("Swap Path is too long. %d > %d ", total_swap_path_length+max_swap_len_suffix, PATH_MAX);
} else {
swap_path = (char *)malloc( total_swap_path_length+max_swap_len_suffix );
strncpy( swap_path, staticConfig.PATH_SWAP.c_str(), swap_path_length );
swap_path = staticConfig.PATH_SWAP;
Debug( 3, "Checking swap path folder: %s", swap_path );
if ( checkSwapPath( swap_path, false ) ) {
// Append the subfolder name /zmswap-m{monitor-id} to the end of swap_path
int ndx = swap_path_length - 1; // Array index of the NULL terminator
snprintf( &(swap_path[ndx]), subfolder1_length, "/zmswap-m%d", monitor->Id() );
Debug( 3, "Checking swap path folder: %s", swap_path.c_str() );
if ( checkSwapPath(swap_path.c_str(), false) ) {
swap_path += stringtf("/zmswap-m%d", monitor->Id());
Debug( 4, "Checking swap path subfolder: %s", swap_path );
if ( checkSwapPath( swap_path, true ) ) {
// Append the subfolder name /zmswap-q{connection key} to the end of swap_path
ndx = swap_path_length+subfolder1_length - 2; // Array index of the NULL terminator
snprintf( &(swap_path[ndx]), subfolder2_length, "/zmswap-q%06d", connkey );
Debug(4, "Checking swap path subfolder: %s", swap_path.c_str());
if ( checkSwapPath(swap_path.c_str(), true) ) {
swap_path += stringtf("/zmswap-q%06d", connkey);
Debug( 4, "Checking swap path subfolder: %s", swap_path );
if ( checkSwapPath( swap_path, true ) ) {
Debug(4, "Checking swap path subfolder: %s", swap_path.c_str());
if ( checkSwapPath(swap_path.c_str(), true) ) {
buffered_playback = true;
}
}
}
if ( !buffered_playback ) {
Error( "Unable to validate swap image path, disabling buffered playback" );
Error("Unable to validate swap image path, disabling buffered playback");
} else {
Debug( 2, "Assigning temporary buffer" );
Debug(2, "Assigning temporary buffer");
temp_image_buffer = new SwapImage[temp_image_buffer_count];
memset( temp_image_buffer, 0, sizeof(*temp_image_buffer)*temp_image_buffer_count );
Debug( 2, "Assigned temporary buffer" );
@ -534,18 +508,18 @@ void MonitorStream::runStream() {
float max_secs_since_last_sent_frame = 10.0; //should be > keep alive amount (5 secs)
while ( !zm_terminate ) {
bool got_command = false;
if ( feof( stdout ) || ferror( stdout ) || !monitor->ShmValid() ) {
if ( feof( stdout ) ) {
Debug(2,"feof stdout");
} else if ( ferror( stdout ) ) {
Debug(2,"ferror stdout");
} else if ( !monitor->ShmValid() ) {
Debug(2,"monitor not valid.... maybe we should wait until it comes back.");
}
if ( feof(stdout) ) {
Debug(2,"feof stdout");
break;
} else if ( ferror(stdout) ) {
Debug(2,"ferror stdout");
break;
} else if ( !monitor->ShmValid() ) {
Debug(2,"monitor not valid.... maybe we should wait until it comes back.");
break;
}
gettimeofday( &now, NULL );
gettimeofday(&now, NULL);
if ( connkey ) {
while(checkCommandQueue()) {
@ -554,7 +528,6 @@ Debug(2, "Have checking command Queue for connkey: %d", connkey );
}
}
//bool frame_sent = false;
if ( buffered_playback && delayed ) {
if ( temp_read_index == temp_write_index ) {
// Go back to live viewing
@ -651,13 +624,13 @@ Debug(2, "Have checking command Queue for connkey: %d", connkey );
}
} // end if should send frame
if ( buffered_playback ) {
if ( buffered_playback && !paused ) {
if ( monitor->shared_data->valid ) {
if ( monitor->image_buffer[index].timestamp->tv_sec ) {
int temp_index = temp_write_index%temp_image_buffer_count;
Debug( 2, "Storing frame %d", temp_index );
Debug(2, "Storing frame %d", temp_index);
if ( !temp_image_buffer[temp_index].valid ) {
snprintf( temp_image_buffer[temp_index].file_name, sizeof(temp_image_buffer[0].file_name), "%s/zmswap-i%05d.jpg", swap_path, temp_index );
snprintf( temp_image_buffer[temp_index].file_name, sizeof(temp_image_buffer[0].file_name), "%s/zmswap-i%05d.jpg", swap_path.c_str(), temp_index );
temp_image_buffer[temp_index].valid = true;
}
memcpy( &(temp_image_buffer[temp_index].timestamp), monitor->image_buffer[index].timestamp, sizeof(temp_image_buffer[0].timestamp) );
@ -697,50 +670,49 @@ Debug(2, "Have checking command Queue for connkey: %d", connkey );
// If we didn't capture above, because frame_mod was bad? Then last_frame_sent will not have a value.
last_frame_sent = now.tv_sec;
Warning( "no last_frame_sent. Shouldn't happen. frame_mod was (%d) frame_count (%d) ", frame_mod, frame_count );
} else if ( (TV_2_FLOAT( now ) - last_frame_sent) > max_secs_since_last_sent_frame ) {
} else if ( (!paused) && ( (TV_2_FLOAT( now ) - last_frame_sent) > max_secs_since_last_sent_frame ) ) {
Error( "Terminating, last frame sent time %f secs more than maximum of %f", TV_2_FLOAT( now ) - last_frame_sent, max_secs_since_last_sent_frame );
break;
}
} // end while
if ( buffered_playback ) {
Debug( 1, "Cleaning swap files from %s", swap_path );
Debug(1, "Cleaning swap files from %s", swap_path.c_str());
struct stat stat_buf;
if ( stat( swap_path, &stat_buf ) < 0 ) {
if ( stat(swap_path.c_str(), &stat_buf) < 0 ) {
if ( errno != ENOENT ) {
Error( "Can't stat '%s': %s", swap_path, strerror(errno) );
Error("Can't stat '%s': %s", swap_path.c_str(), strerror(errno));
}
} else if ( !S_ISDIR(stat_buf.st_mode) ) {
Error( "Swap image path '%s' is not a directory", swap_path );
Error("Swap image path '%s' is not a directory", swap_path.c_str());
} else {
char glob_pattern[PATH_MAX] = "";
snprintf( glob_pattern, sizeof(glob_pattern), "%s/*.*", swap_path );
snprintf(glob_pattern, sizeof(glob_pattern), "%s/*.*", swap_path.c_str());
glob_t pglob;
int glob_status = glob( glob_pattern, 0, 0, &pglob );
int glob_status = glob(glob_pattern, 0, 0, &pglob);
if ( glob_status != 0 ) {
if ( glob_status < 0 ) {
Error( "Can't glob '%s': %s", glob_pattern, strerror(errno) );
Error("Can't glob '%s': %s", glob_pattern, strerror(errno));
} else {
Debug( 1, "Can't glob '%s': %d", glob_pattern, glob_status );
Debug(1, "Can't glob '%s': %d", glob_pattern, glob_status);
}
} else {
for ( unsigned int i = 0; i < pglob.gl_pathc; i++ ) {
if ( unlink( pglob.gl_pathv[i] ) < 0 ) {
Error( "Can't unlink '%s': %s", pglob.gl_pathv[i], strerror(errno) );
if ( unlink(pglob.gl_pathv[i]) < 0 ) {
Error("Can't unlink '%s': %s", pglob.gl_pathv[i], strerror(errno));
}
}
}
globfree( &pglob );
if ( rmdir( swap_path ) < 0 ) {
Error( "Can't rmdir '%s': %s", swap_path, strerror(errno) );
if ( rmdir(swap_path.c_str()) < 0 ) {
Error( "Can't rmdir '%s': %s", swap_path.c_str(), strerror(errno) );
}
} // end if checking for swap_path
} // end if buffered_playback
if ( swap_path ) free( swap_path );
closeComms();
}
} // end MonitorStream::runStream
void MonitorStream::SingleImage( int scale ) {
int img_buffer_size = 0;

View File

@ -39,48 +39,49 @@ StreamBase::~StreamBase() {
closeComms();
}
bool StreamBase::loadMonitor( int monitor_id ) {
if ( !(monitor = Monitor::Load( monitor_id, false, Monitor::QUERY )) ) {
Fatal( "Unable to load monitor id %d for streaming", monitor_id );
return( false );
bool StreamBase::loadMonitor(int monitor_id) {
if ( !(monitor = Monitor::Load(monitor_id, false, Monitor::QUERY)) ) {
Error("Unable to load monitor id %d for streaming", monitor_id);
return false;
}
if ( ! monitor->connect() ) {
Fatal( "Unable to connect to monitor id %d for streaming", monitor_id );
return( false );
Error("Unable to connect to monitor id %d for streaming", monitor_id);
return false;
}
return( true );
return true;
}
bool StreamBase::checkInitialised() {
if ( !monitor ) {
Fatal( "Cannot stream, not initialised" );
return( false );
return false;
}
return( true );
return true;
}
void StreamBase::updateFrameRate( double fps ) {
void StreamBase::updateFrameRate(double fps) {
base_fps = fps;
effective_fps = (base_fps*abs(replay_rate))/ZM_RATE_BASE;
frame_mod = 1;
Debug( 3, "FPS:%.2f, MXFPS:%.2f, BFPS:%.2f, EFPS:%.2f, FM:%d", fps, maxfps, base_fps, effective_fps, frame_mod );
Debug(3, "FPS:%.2f, MXFPS:%.2f, BFPS:%.2f, EFPS:%.2f, FM:%d", fps, maxfps, base_fps, effective_fps, frame_mod);
// Min frame repeat?
while( effective_fps > maxfps ) {
effective_fps /= 2.0;
frame_mod *= 2;
Debug( 3, "aEFPS:%.2f, aFM:%d", effective_fps, frame_mod );
Debug(3, "EffectiveFPS:%.2f, FrameMod:%d", effective_fps, frame_mod);
}
}
bool StreamBase::checkCommandQueue() {
if ( sd >= 0 ) {
CmdMsg msg;
memset( &msg, 0, sizeof(msg) );
int nbytes = recvfrom( sd, &msg, sizeof(msg), MSG_DONTWAIT, 0, 0 );
memset(&msg, 0, sizeof(msg));
int nbytes = recvfrom(sd, &msg, sizeof(msg), MSG_DONTWAIT, 0, 0);
if ( nbytes < 0 ) {
if ( errno != EAGAIN ) {
Fatal( "recvfrom(), errno = %d, error = %s", errno, strerror(errno) );
Error("recvfrom(), errno = %d, error = %s", errno, strerror(errno));
return false;
}
}
//else if ( (nbytes != sizeof(msg)) )
@ -88,14 +89,14 @@ bool StreamBase::checkCommandQueue() {
//Error( "Partial message received, expected %d bytes, got %d", sizeof(msg), nbytes );
//}
else {
Debug(2, "Message length is (%d)", nbytes );
processCommand( &msg );
return( true );
Debug(2, "Message length is (%d)", nbytes);
processCommand(&msg);
return true;
}
} else {
Warning("No sd in checkCommandQueue, comms not open?");
}
return( false );
return false;
}
Image *StreamBase::prepareImage( Image *image ) {
@ -152,18 +153,18 @@ Image *StreamBase::prepareImage( Image *image ) {
if ( mag != ZM_SCALE_BASE ) {
if ( act_mag != ZM_SCALE_BASE ) {
Debug( 3, "Magnifying by %d", mag );
Debug(3, "Magnifying by %d", mag);
if ( !image_copied ) {
static Image copy_image;
copy_image.Assign( *image );
copy_image.Assign(*image);
image = &copy_image;
image_copied = true;
}
image->Scale( mag );
image->Scale(mag);
}
}
Debug( 3, "Real image width = %d, height = %d", image->Width(), image->Height() );
Debug(3, "Real image width = %d, height = %d", image->Width(), image->Height());
if ( disp_image_width < virt_image_width || disp_image_height < virt_image_height ) {
static Box last_crop;
@ -222,23 +223,23 @@ Image *StreamBase::prepareImage( Image *image ) {
last_x = x;
last_y = y;
return( image );
return image;
}
bool StreamBase::sendTextFrame( const char *frame_text ) {
Debug( 2, "Sending text frame '%s'", frame_text );
Debug(2, "Sending text frame '%s'", frame_text);
Image image( monitor->Width(), monitor->Height(), monitor->Colours(), monitor->SubpixelOrder() );
image.Annotate( frame_text, image.centreCoord( frame_text ) );
Image image(monitor->Width(), monitor->Height(), monitor->Colours(), monitor->SubpixelOrder());
image.Annotate(frame_text, image.centreCoord(frame_text));
if ( scale != 100 ) {
image.Scale( scale );
image.Scale(scale);
}
#if HAVE_LIBAVCODEC
if ( type == STREAM_MPEG ) {
if ( !vid_stream ) {
vid_stream = new VideoStream( "pipe:", format, bitrate, effective_fps, image.Colours(), image.SubpixelOrder(), image.Width(), image.Height() );
fprintf( stdout, "Content-type: %s\r\n\r\n", vid_stream->MimeType() );
vid_stream = new VideoStream("pipe:", format, bitrate, effective_fps, image.Colours(), image.SubpixelOrder(), image.Width(), image.Height());
fprintf(stdout, "Content-type: %s\r\n\r\n", vid_stream->MimeType());
vid_stream->OpenStream();
}
/* double pts = */ vid_stream->EncodeFrame( image.Buffer(), image.Size() );
@ -248,30 +249,29 @@ bool StreamBase::sendTextFrame( const char *frame_text ) {
static unsigned char buffer[ZM_MAX_IMAGE_SIZE];
int n_bytes = 0;
image.EncodeJpeg( buffer, &n_bytes );
image.EncodeJpeg(buffer, &n_bytes);
fprintf( stdout, "--ZoneMinderFrame\r\n" );
fprintf( stdout, "Content-Length: %d\r\n", n_bytes );
fprintf( stdout, "Content-Type: image/jpeg\r\n\r\n" );
if ( fwrite( buffer, n_bytes, 1, stdout ) != 1 ) {
Error( "Unable to send stream text frame: %s", strerror(errno) );
return( false );
fputs("--ZoneMinderFrame\r\nContent-Type: image/jpeg\r\n\r\n", stdout);
fprintf(stdout, "Content-Length: %d\r\n", n_bytes);
if ( fwrite(buffer, n_bytes, 1, stdout) != 1 ) {
Error("Unable to send stream text frame: %s", strerror(errno));
return false;
}
fprintf( stdout, "\r\n\r\n" );
fflush( stdout );
fputs("\r\n\r\n",stdout);
fflush(stdout);
}
last_frame_sent = TV_2_FLOAT( now );
return( true );
last_frame_sent = TV_2_FLOAT(now);
return true;
}
void StreamBase::openComms() {
if ( connkey > 0 ) {
unsigned int length = snprintf( sock_path_lock, sizeof(sock_path_lock), "%s/zms-%06d.lock", staticConfig.PATH_SOCKS.c_str(), connkey);
unsigned int length = snprintf(sock_path_lock, sizeof(sock_path_lock), "%s/zms-%06d.lock", staticConfig.PATH_SOCKS.c_str(), connkey);
if ( length >= sizeof(sock_path_lock) ) {
Warning("Socket lock path was truncated.");
}
Debug( 1, "Trying to open the lock on %s", sock_path_lock );
Debug(1, "Trying to open the lock on %s", sock_path_lock);
lock_fd = open(sock_path_lock, O_CREAT|O_WRONLY, S_IRUSR | S_IWUSR);
if ( lock_fd <= 0 ) {
@ -285,46 +285,46 @@ void StreamBase::openComms() {
Debug( 1, "We have obtained a lock on %s fd: %d", sock_path_lock, lock_fd);
}
sd = socket( AF_UNIX, SOCK_DGRAM, 0 );
sd = socket(AF_UNIX, SOCK_DGRAM, 0);
if ( sd < 0 ) {
Fatal( "Can't create socket: %s", strerror(errno) );
Fatal("Can't create socket: %s", strerror(errno));
} else {
Debug(1, "Have socket %d", sd );
Debug(1, "Have socket %d", sd);
}
length = snprintf( loc_sock_path, sizeof(loc_sock_path), "%s/zms-%06ds.sock", staticConfig.PATH_SOCKS.c_str(), connkey );
length = snprintf(loc_sock_path, sizeof(loc_sock_path), "%s/zms-%06ds.sock", staticConfig.PATH_SOCKS.c_str(), connkey);
if ( length >= sizeof(loc_sock_path) ) {
Warning("Socket path was truncated.");
length = sizeof(loc_sock_path)-1;
}
// Unlink before bind, in case it already exists
unlink( loc_sock_path );
unlink(loc_sock_path);
if ( sizeof(loc_addr.sun_path) < length ) {
Error("Not enough space %d in loc_addr.sun_path for socket file %s", sizeof(loc_addr.sun_path), loc_sock_path );
Error("Not enough space %d in loc_addr.sun_path for socket file %s", sizeof(loc_addr.sun_path), loc_sock_path);
}
strncpy( loc_addr.sun_path, loc_sock_path, sizeof(loc_addr.sun_path) );
strncpy(loc_addr.sun_path, loc_sock_path, sizeof(loc_addr.sun_path));
loc_addr.sun_family = AF_UNIX;
Debug(3, "Binding to %s", loc_sock_path );
if ( bind( sd, (struct sockaddr *)&loc_addr, strlen(loc_addr.sun_path)+sizeof(loc_addr.sun_family)+1 ) < 0 ) {
Fatal( "Can't bind: %s", strerror(errno) );
Debug(3, "Binding to %s", loc_sock_path);
if ( bind(sd, (struct sockaddr *)&loc_addr, strlen(loc_addr.sun_path)+sizeof(loc_addr.sun_family)+1) < 0 ) {
Fatal("Can't bind: %s", strerror(errno));
}
snprintf( rem_sock_path, sizeof(rem_sock_path), "%s/zms-%06dw.sock", staticConfig.PATH_SOCKS.c_str(), connkey );
strncpy( rem_addr.sun_path, rem_sock_path, sizeof(rem_addr.sun_path)-1 );
snprintf(rem_sock_path, sizeof(rem_sock_path), "%s/zms-%06dw.sock", staticConfig.PATH_SOCKS.c_str(), connkey);
strncpy(rem_addr.sun_path, rem_sock_path, sizeof(rem_addr.sun_path)-1);
rem_addr.sun_family = AF_UNIX;
} // end if connKey > 0
Debug(2, "comms open" );
Debug(2, "comms open");
} // end void StreamBase::openComms()
void StreamBase::closeComms() {
if ( connkey > 0 ) {
if ( sd >= 0 ) {
close( sd );
close(sd);
sd = -1;
}
if ( loc_sock_path[0] ) {
unlink( loc_sock_path );
unlink(loc_sock_path);
}
if ( lock_fd > 0 ) {
close(lock_fd); //close it rather than unlock it incase it got deleted.
@ -332,4 +332,4 @@ void StreamBase::closeComms() {
//unlink(sock_path_lock);
}
}
}
} // end void StreamBase::closeComms