sender.cpp

voudp - VoIP for NetBSD

//
// Copyright 2004 Alan Post
//
// This file is part of voudp.
//
// voudp 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 2 of the License, or (at your option) any later
// version.
//
// voudp 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
// voudp; if not, write to the Free Software Foundation, Inc., 59 Temple Place,
// Suite 330, Boston, MA  02111-1307  USA
//

//
// XXX:  use sendto(2) instead of send(2), and have only one sending socket
//

#include "sender.h"
#include "constants.h"
#include "die.h"

#include <stdlib.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <fcntl.h>
#include <assert.h>
#include <stdio.h>
#include <unistd.h>
#include <strings.h>
#include <string.h>
#include <sys/time.h>
#include <map>
#include <err.h>

#include "speex.h"

static in_addr_t lookup( const std::string &host )
{
    in_addr_t addr;
    struct hostent *he;

    addr = inet_addr( host.c_str());
    if ( addr != (in_addr_t)(-1)) return addr;

    he = gethostbyname( host.c_str());
    if ( he == NULL )
    {
        herror( host.c_str());
        exit( EXIT_FAILURE );
    }
    assert( he->h_addr != NULL );
    memcpy( &addr, he->h_addr, sizeof( addr ));

    return addr;
}

static int hookup( const std::string &host, const int &port )
{
    struct sockaddr_in sa;
    bzero( &sa, sizeof( sa ));
#ifndef LINUX
    sa.sin_len = sizeof( sa );
#endif
    sa.sin_family = AF_INET;
    sa.sin_port = htons( port );
    sa.sin_addr.s_addr = lookup( host );

    int fd = socket( PF_INET, SOCK_DGRAM, 0 );
    if ( fd == -1 ) die( "socket" );
    if ( 0 != connect( fd, (struct sockaddr*) &sa, sizeof( sa )))
        die( "connect" );
    return fd;
}

class packet_t
{
public:
    packet_t( const client_id_t &cid, const recipient_info_t &info )
    {
        buf = 0;
        send_fd = -1;
        frames_in_buffer = 0;
        n_p_sent = 0;
        n_f_sent = 0;

        use_info( cid, info );
    }

    void use_info( const client_id_t &cid, const recipient_info_t &info )
    {
        n_f_sent -= frames_in_buffer;
        delete[] buf;
        if ( send_fd != -1 ) close( send_fd );

        u_int32_t n_cid = htonl( cid );

        buf = new u_int8_t[ info.max_packet_len ];
        memcpy( buf, &n_cid, sizeof( n_cid ));
        buf_len = info.max_packet_len;
        buf_pos = sizeof( n_cid );
        frames_in_buffer = 0;
        max_frames =   info.max_buf_ms * ( samples_per_sec / 1000 )
                     / samples_per_frame;
        send_fd = hookup( info.host, info.port );
    }

    ~packet_t()
    {
        send_shutdown_packet();
        delete[] buf;
        if ( -1 == close( send_fd )) warn( "Closing send socket" );
    }

    void push_frame( const frame_id_t  &fid,
                     const u_int8_t    *frame,
                     const frame_len_t &frame_len )
    {
        if ( ! can_fit( frame_len ))
        {
            if ( ! send_packet()) return;
        }
        n_f_sent++;

        u_int32_t n_fid = htonl( fid );
        memcpy( buf + buf_pos, &n_fid, sizeof( n_fid ));
        buf_pos += sizeof( n_fid );
        memcpy( buf + buf_pos, &frame_len, sizeof( frame_len ));
        buf_pos += sizeof( frame_len );
        memcpy( buf + buf_pos, frame, frame_len );
        buf_pos += frame_len;
        frames_in_buffer++;

        //
        // If output buffer is full, send the packet immediately.
        //
        if ( ! can_fit( 1 )) { (void) send_packet(); }
    }
private:
    bool can_fit( const frame_len_t &frame_len )
    {
        return       ( buf_pos + frame_header_size + frame_len )
                  <= buf_len
               && frames_in_buffer < max_frames;
    }

    bool send_packet()
    {
        assert( frames_in_buffer > 0 );
        if ( -1 == send( send_fd, buf, buf_pos, 0 ))
        {
            warn( "send" );
            return false;
        }
        n_p_sent++;
        buf_pos = sizeof( client_id_t );
        frames_in_buffer = 0;
        return true;
    }

    void send_shutdown_packet()
    {
        if ( -1 == send( send_fd, buf, sizeof( client_id_t ), 0 ))
            warn( "send" );
    }


    u_int8_t *buf;
    ssize_t buf_len;
    ssize_t buf_pos;
    ssize_t frames_in_buffer;
    ssize_t max_frames;
    frame_id_t next_fid;
    int send_fd;

    u_int n_p_sent;
    u_int n_f_sent;
};

static client_id_t get_client_id( void )
{
    u_int32_t id, sec, usec;
    struct timeval now;
    if ( 0 != gettimeofday( &now, NULL )) die( "gettimeofday" );
    sec = now.tv_sec;
    usec = now.tv_usec;
    id = (sec << 8) | (usec >> 24);
    return id;
}

void
Sender::push_frame( const float *frame )
{
    frame_id_t fid = next_fid;
    next_fid++;

    if ( actives.empty()) return;

    SpeexBits *bits = (SpeexBits*) speex_bits;
    speex_bits_reset( bits );
    speex_encode( enc_state, (float*)frame, bits );
    u_int8_t frame_len = speex_bits_nbytes( bits );
    u_int8_t *frame_bytes = new u_int8_t[ frame_len ];
    frame_len = speex_bits_write( bits, (char*)frame_bytes, frame_len );

    for ( actives_t::iterator i = actives.begin();
          i != actives.end(); ++i )
    {
        (*i).second->push_frame( fid, frame_bytes, frame_len );
    }
    delete[] frame_bytes;
}

Sender::Sender()
{
    cid = get_client_id();
    next_fid = 0;

    speex_bits = new SpeexBits;
    speex_bits_init( (SpeexBits*)speex_bits );

    enc_state = speex_encoder_init( &speex_nb_mode );
    {
        int tmp = 7;
        speex_encoder_ctl( enc_state, SPEEX_SET_QUALITY, &tmp );
    }
}

Sender::~Sender()
{
    for ( infos_t::iterator i = infos.begin();
          i != infos.end(); )
    {
        std::string name = (*i).first;
        ++i;
        delete_recipient( name.c_str());
    }
    assert( infos.empty());
    assert( actives.empty());

    speex_bits_destroy( (SpeexBits*)speex_bits );
    delete (SpeexBits*)speex_bits;
    speex_encoder_destroy( enc_state );
}

void
Sender::set_recipient_info( const std::string      &name,
                            const recipient_info_t &info )
{
    infos[ name ] = info;
    if ( actives.count( name ))
    {
        actives[ name ]->use_info( cid, info );
    }
}

bool
Sender::get_recipient_info( const std::string &name, recipient_info_t &info )
{
    if ( ! infos.count( name )) return false;
    info = infos[ name ];
    return true;
}
    
bool
Sender::start_recipient( const std::string &name )
{
    if ( ! infos.count( name )) return false;
    if ( actives.count( name )) return true;
    actives[ name ] = new packet_t( cid, infos[ name ]);
    return true;
}

bool
Sender::stop_recipient( const std::string &name )
{
    if ( ! infos.count( name )) return false;
    if ( ! actives.count( name )) return true;
    delete actives[ name ];
    actives.erase( name );
    return true;
}

bool
Sender::is_recipient_active( const std::string &name )
{
    return 0 < actives.count( name );
}

void
Sender::delete_recipient( const std::string &name )
{
    if ( ! infos.count( name )) return;
    if ( actives.count( name )) stop_recipient( name );
    infos.erase( name );
}

Sender::infos_t::const_iterator
Sender::infos_begin() const { return infos.begin(); }

Sender::infos_t::const_iterator
Sender::infos_end() const { return infos.end(); }