Mickey_Libevent_WhyNoWorkie

struct Connection
{
   Connection(): bev(NULL), isBusy( false ), needsService( false ) {}
   bufferevent* bev;
   char buffer[MAX_PATH];
   bool isBusy;// should be a thread atomic, but whatever
   bool needsService;

   void  SendBuffer()
   {
      assert ( isBusy == false );
      isBusy = true;
      struct evbuffer* outputBuffer = bufferevent_get_output( bev );
      int length = strlen( buffer );
      evbuffer_add( outputBuffer, buffer, length );

      needsService = false;
      isBusy = false;
   }
};

//*************************************************
//*************************************************

vector< Connection* > connections;  /// global... testing only... needs threading protections

//*************************************************
//*************************************************

class MY_THREAD : public THREAD_CLASS
{
    int Num;
protected:
    void MainThreadLoop ()
    {
      // major threading issue here.. needs protections
      vector< Connection* >::iterator it = connections.begin();
      while( it != connections.end() )
      {
         Connection* conn = *it;
         vector< Connection* >::iterator current = it++;
         if( conn == NULL )
         {
            connections.erase( current );
            continue;
         }
         if( conn->isBusy == false )
         {
            if( conn->needsService== true )
            {
               conn->SendBuffer();
            }
         }
      }
    }
public:
    MY_THREAD (bool CreateSuspended = true) : THREAD_CLASS (CreateSuspended)
    {
        Num = 0;
    }
};

//-------------------------------------------------------------------

static void
echo_read_cb(struct bufferevent *bufferEventObj, void *ctx)
{
   Connection* conn = static_cast< Connection * >( ctx );
   assert ( conn->isBusy == false );

   vector< Connection* >::iterator it =  connections.begin();
   bool found = false;
   while( it != connections.end() )
   {
      if( conn == *it )
      {
         found = true;
         break;
      }
      it++;
   }
   if( found == false )
      assert( 0 );

   conn->isBusy = true;
   int numBytesReceived = bufferevent_read( bufferEventObj, conn->buffer, sizeof( conn->buffer ) );
   if( numBytesReceived > 0 )
      conn->buffer[ numBytesReceived ] = 0;
   conn->isBusy = false;
   conn->needsService = true;
}

static void
echo_event_cb(struct bufferevent *bev, short events, void *ctx)
{
        if (events & BEV_EVENT_ERROR)
                perror("Error from bufferevent");
        if (events & (BEV_EVENT_EOF | BEV_EVENT_ERROR)) {
                bufferevent_free(bev);
            Connection* conn = static_cast< Connection * >( ctx );
            vector< Connection* >::iterator it = connections.begin();
            while( it != connections.end() )
            {
               if( *it == conn )
               {
                  *it = NULL;
               }
               it++;
            }

        }
}

static void
accept_conn_cb(struct evconnlistener *listener,
    evutil_socket_t fd, struct sockaddr *address, int socklen,
    void *ctx)
{
        /* We got a new connection! Set up a bufferevent for it. */
        struct event_base *base = evconnlistener_get_base(listener);
        struct bufferevent *bev = bufferevent_socket_new(
                base, fd, BEV_OPT_CLOSE_ON_FREE);

        Connection* conn = new Connection;
        conn->bev = bev;

        strcpy( conn->buffer, "ready" );
        conn->needsService = true;
        connections.push_back( conn );

        bufferevent_setcb( bev, echo_read_cb, NULL, echo_event_cb, conn );

        bufferevent_enable(bev, EV_READ|EV_WRITE);
}

static void
accept_error_cb(struct evconnlistener *listener, void *ctx)
{
        struct event_base *base = evconnlistener_get_base(listener);
        int err = EVUTIL_SOCKET_ERROR();
        fprintf(stderr, "Got an error %d (%s) on the listener. "
                "Shutting down.\n", err, evutil_socket_error_to_string(err));

        event_base_loopexit(base, NULL);
}

int
main(int argc, char **argv)
{
   WSADATA WsaData;
    WSAStartup( MAKEWORD(2,2), &WsaData );
        struct event_base *base;
        struct evconnlistener *listener;
        struct sockaddr_in sin;

        int port = 5555;

        if (argc > 1) {
                port = atoi(argv[1]);
        }
        if (port<=0 || port>65535) {
                puts("Invalid port");
                return 1;
        }

        base = event_base_new();
        if (!base) {
                puts("Couldn't open event base");
                return 1;
        }

        /* Clear the sockaddr before using it, in case there are extra
         * platform-specific fields that can mess us up. */
        memset(&sin, 0, sizeof(sin));/* This is an INET address */
        sin.sin_family = AF_INET;/* Listen on 0.0.0.0 */
        sin.sin_addr.s_addr = htonl(0);/* Listen on the given port. */
        sin.sin_port = htons(port);

        listener = evconnlistener_new_bind(base, accept_conn_cb, NULL,
            LEV_OPT_CLOSE_ON_FREE|LEV_OPT_REUSEABLE, -1,
            (struct sockaddr*)&sin, sizeof(sin));
        if (!listener) {
                perror("Couldn't create listener");
                return 1;
        }
        evconnlistener_set_error_cb(listener, accept_error_cb);

        MY_THREAD mainThread;
        mainThread.SetAlarm( 30 );// fire every 30 millisecs

        mainThread.StartThread();

        event_base_dispatch(base);
        return 0;
}