attribute.c body.c header.c io.c mbx_pop.c mbx_unix.c

 	mbx_unixscan.c message.c include/private/message0.h
 	include/public/io.h
cleanup, finish the POP client code hopefully.

@@ -40,8 +40,7 @@ attribute_destroy (attribute_t *pattr)
 {
   if (pattr && *pattr)
     {
-      attribute_t attr = *pattr;
-      free (attr);
+      free (*pattr);
       /* loose the link */
       *pattr = NULL;
     }

@@ -58,11 +58,8 @@ body_destroy (body_t *pbody, void *owner)
       if (body->owner == owner)
 	{
 	  if (body->file)
-	    {
-	      fclose (body->file);
-	      body->file = NULL;
-	    }
-	  free (body->filename); body->filename = NULL;
+	    fclose (body->file);
+	  free (body->filename);
 	  stream_destroy (&(body->stream), body);
 	}
       *pbody = NULL;

@@ -53,14 +53,12 @@ struct _header
 
   /* owner ? */
   void *owner;
-  int ref_count;
 };
 
 int
 header_create (header_t *ph, const char *blurb, size_t len, void *owner)
 {
   header_t h;
-  int status;
   h = calloc (1, sizeof (*h));
   if (h == NULL)
     return ENOMEM;
@@ -68,15 +66,8 @@ header_create (header_t *ph, const char *blurb, size_t len, void *owner)
 
   header_parse (h, (char *)blurb, len);
 
-  status = stream_create (&(h->stream), MU_STREAM_READ|MU_STREAM_WRITE, h);
-  if (status != 0)
-    return status;
-
-  stream_set_read  (h->stream, header_read, h);
-  stream_set_write (h->stream, header_write, h);
-
   *ph = h;
-  return status;
+  return 0;
 }
 
 void
@@ -86,16 +77,11 @@ header_destroy (header_t *ph, void *owner)
     {
       header_t h = *ph;
 
-      /* if destroy is call always decremente */
-      h->ref_count--;
-
       /* can we destroy ? */
-      if ((h->owner && h->owner == owner) ||
-	  (h->owner == NULL && h->ref_count <= 0))
+      if (h->owner == owner)
 	{
 	  /* io */
 	  stream_destroy (&(h->stream), h);
-
 	  free (h->hdr);
 	  free (h->blurb);
 	  free (h);
@@ -427,6 +413,14 @@ header_get_stream (header_t header, stream_t *pstream)
 {
   if (header == NULL || pstream == NULL)
     return EINVAL;
+  if (header->stream == NULL)
+    {
+      int status = stream_create (&(header->stream), MU_STREAM_RDWR, header);
+      if (status != 0)
+	return status;
+      stream_set_read  (header->stream, header_read, header);
+      stream_set_write (header->stream, header_write, header);
+    }
   *pstream = header->stream;
   return 0;
 }

@@ -46,11 +46,9 @@ struct _message
   stream_t stream;
   body_t body;
   attribute_t attribute;
-  size_t num;
 
   /* who is the owner */
   void *owner;
-  int ref_count;
 
   event_t event;
   size_t event_num;

@@ -71,23 +71,25 @@ extern int stream_set_read __P ((stream_t,
 						    size_t, off_t, size_t *)),
 				 void *owner));
 
-extern int stream_write __P ((stream_t, const char *, size_t, off_t, size_t *));
+extern int stream_write __P ((stream_t, const char *, size_t,
+			      off_t, size_t *));
 extern int stream_set_write __P ((stream_t,
 				  int (*_write) __P ((stream_t, const char *,
 						size_t, off_t,
 						size_t *)),
 				  void *owner));
 
-extern int stream_get_flags __P ((stream_t , int *flags));
+extern int stream_get_flags __P ((stream_t, int *pflags));
+extern int stream_set_flags __P ((stream_t, int flags, void *owner));
 
 /* misc */
 extern int file_stream_create	__P ((stream_t *stream, const char *filename,
 				      int flags));
-extern int encoder_stream_create	__P ((stream_t *stream, stream_t iostream,
-				      const char *encoding));
-extern int decoder_stream_create	__P ((stream_t *stream, stream_t iostream,
-				      const char *encoding));
-extern int tcp_stream_create	__P ((stream_t *stream));
+extern int encoder_stream_create  __P ((stream_t *stream, stream_t iostream,
+					const char *encoding));
+extern int decoder_stream_create  __P ((stream_t *stream, stream_t iostream,
+					const char *encoding));
+extern int tcp_stream_create __P ((stream_t *stream));
 
 #ifdef __cplusplus
 }

@@ -42,11 +42,12 @@ stream_destroy (stream_t *pstream, void *owner)
    if (pstream && *pstream)
     {
       stream_t stream = *pstream;
-      if (!(stream->flags & MU_STREAM_NO_CHECK) && stream->owner != owner)
-	return;
-      if (stream->_destroy)
-	stream->_destroy (stream);
-	free (stream);
+      if ((stream->flags & MU_STREAM_NO_CHECK) || stream->owner == owner)
+	{
+	  if (stream->_destroy)
+	    stream->_destroy (stream);
+	  free (stream);
+	}
       *pstream = NULL;
     }
 }
@@ -193,3 +194,14 @@ stream_get_flags (stream_t stream, int *pfl)
   *pfl = stream->flags;
   return 0;
 }
+
+int
+stream_set_flags (stream_t stream, int fl, void *owner)
+{
+  if (stream == NULL)
+    return EINVAL;
+  if (stream->owner != owner)
+    return EACCES;
+  stream->flags = fl;
+  return 0;
+}

@@ -111,7 +111,7 @@ typedef struct _mailbox_pop_message * mailbox_pop_message_t;
 struct _mailbox_pop_message
 {
   bio_t bio;
-  int started, inbody;
+  int inbody;
   size_t num;
   off_t body_size;
   message_t message;
@@ -230,10 +230,33 @@ mailbox_pop_create (mailbox_t *pmbox, const char *name)
 }
 
 static void
-mailbox_pop_destroy (mailbox_t *mbox)
+mailbox_pop_destroy (mailbox_t *pmbox)
 {
-  (void)mbox;
-  return;
+  if (pmbox && *pmbox)
+    {
+      if ((*pmbox)->data)
+	{
+	  mailbox_pop_data_t mpd = (*pmbox)->data;
+	  size_t i;
+	  for (i = 0; i < mpd->pmessages_count; i++)
+	    {
+	      if (mpd->pmessages[i])
+		{
+		  bio_destroy (&(mpd->pmessages[i]->bio));
+		  message_destroy (&(mpd->pmessages[i]->message),
+				   mpd->pmessages[i]);
+		}
+	      free (mpd->pmessages[i]);
+	    }
+	  free (mpd->pmessages);
+	  free (mpd);
+	}
+      free ((*pmbox)->name);
+      free ((*pmbox)->event);
+      if ((*pmbox)->url)
+	url_destroy (&((*pmbox)->url));
+      *pmbox = NULL;
+    }
 }
 
 static struct termios stored_settings;
@@ -471,15 +494,45 @@ static int
 mailbox_pop_close (mailbox_t mbox)
 {
   mailbox_pop_data_t mpd;
-  //mailbox_pop_message_t mpm;
+  void *func = mailbox_pop_close;
+  int status;
+  bio_t bio;
 
   if (mbox == NULL || (mpd = mbox->data) == NULL)
     return EINVAL;
 
+  if (mpd->func && mpd->func != func)
+    return EBUSY;
+
+  mpd->func = func;
+  bio = mpd->bio;
+
   if (mpd->fd != -1)
-    close (mpd->fd);
-  mpd->fd = -1;
+    {
+      switch (mpd->state)
+	{
+	case 0:
+	  bio->len = sprintf (bio->buffer, "QUIT\r\n");
+	  bio->ptr = bio->buffer;
+	  mpd->state = 1;
+	case 1:
+	  status = bio_write (mpd->bio);
+	  if (status != 0)
+	    {
+	       if (status != EAGAIN && status != EINTR)
+		 {
+		   mpd->func = mpd->id = NULL;
+		   mpd->state = 0;
+		 }
+	       return status;
+	    }
+	  close (mpd->fd);
+	  mpd->fd = -1;
+	}
+    }
 
+  mpd->func = mpd->id = NULL;
+  mpd->state = 0;
   return 0;
 }
 
@@ -619,8 +672,8 @@ mailbox_pop_get_message (mailbox_t mbox, size_t msgno, message_t *pmsg)
 	    mpd->mpm = NULL;
 	    return ERANGE;
 	  }
-	mpd->state = 5;
       }
+      mpd->state = 5;
       /* get the header */
     case 5:
       {
@@ -646,16 +699,12 @@ mailbox_pop_get_message (mailbox_t mbox, size_t msgno, message_t *pmsg)
 
 	    /* our ticket out */
 	    if (bio->buffer[0] == '\0')
-	      {
-		mpd->mpm->bio->buffer[mpd->mpm->bio->maxlen + 1] = '\0';
-		break;
-	      }
+	      break;
 
-	    nread = (bio->nl) ? bio->nl - bio->buffer :
-	      bio->ptr - bio->buffer + 1;
+	    nread = bio->nl - bio->buffer;
 
 	    tbuf = realloc (mpd->mpm->bio->buffer,
-			    mpd->mpm->bio->maxlen + nread + 1);
+			    mpd->mpm->bio->maxlen + nread);
 	    if (tbuf == NULL)
 	      {
 		mpd->func = mpd->id = NULL;
@@ -667,7 +716,7 @@ mailbox_pop_get_message (mailbox_t mbox, size_t msgno, message_t *pmsg)
 		return ENOMEM;
 	      }
 	    else
-	      mpd->mpm->bio->buffer = (void *)tbuf;
+	      mpd->mpm->bio->buffer = tbuf;
 	    memcpy (mpd->mpm->bio->buffer + mpd->mpm->bio->maxlen,
 		    bio->buffer, nread);
 	    mpd->mpm->bio->maxlen += nread;
@@ -748,12 +797,12 @@ mailbox_pop_get_message (mailbox_t mbox, size_t msgno, message_t *pmsg)
       }
     stream_set_read (stream, mailbox_pop_readstream, mpd->mpm);
     stream_set_fd (stream, mailbox_pop_getfd, mpd->mpm);
+    stream_set_flags (stream, mpd->flags, mpd->mpm);
     body_set_size (body, mailbox_pop_body_size, mpd->mpm);
     //body_set_lines (body, mailbox_pop_body_lines, mpd->mpm);
     body_set_stream (body, stream, mpd->mpm);
   }
 
-
   /* add it to the list */
   {
     mailbox_pop_message_t *m ;
@@ -888,19 +937,74 @@ mailbox_pop_expunge (mailbox_t mbox)
   mailbox_pop_data_t mpd;
   size_t i;
   attribute_t attr;
+  bio_t bio;
+  int status;
+  void *func = mailbox_pop_expunge;
 
   if (mbox == NULL ||
       (mpd = (mailbox_pop_data_t) mbox->data) == NULL)
     return EINVAL;
-  for (i = 0; i < mpd->messages_count; i++)
+
+  /* busy ? */
+  if (mpd->func && mpd->func != func)
+    return EBUSY;
+
+  mpd->func = func;
+  bio = mpd->bio;
+
+  for (i = (int)mpd->id; i < mpd->messages_count; mpd->id = (void *)++i)
     {
-      if (message_get_attribute (mpd->pmessages[i]->message, &attr) != 0)
+      if (message_get_attribute (mpd->pmessages[i]->message, &attr) == 0)
 	{
+	  /* send DELETE */
 	  if (attribute_is_deleted (attr))
 	    {
-	    }
-	}
-    }
+	      switch (mpd->state)
+		{
+		case 0:
+		  bio->len = sprintf (bio->buffer, "DELE %d\r\n",
+				      mpd->pmessages[i]->num);
+		  bio->ptr = bio->buffer;
+		  mpd->state = 1;
+		case 1:
+		  status = bio_write (bio);
+		  if (status != 0)
+		    {
+		      if (status != EAGAIN && status != EINTR)
+			{
+			  mpd->func = mpd->id = NULL;
+			  mpd->state = 0;
+			  fprintf(stderr, "PROBLEM write %d\n", status);
+			}
+		      return status;
+		    }
+		  mpd->state = 2;
+		case 2:
+		  status = bio_readline (bio);
+		  if (status != 0)
+		    {
+		      if (status != EAGAIN && status != EINTR)
+			{
+			  mpd->func = mpd->id = NULL;
+			  mpd->state = 0;
+			  fprintf(stderr, "PROBLEM readline %d\n", status);
+			}
+		      return status;
+		    }
+		  if (strncasecmp (bio->buffer, "+OK", 3) != 0)
+		    {
+		      mpd->func = mpd->id = NULL;
+		      mpd->state = 0;
+			  fprintf(stderr, "PROBLEM strcmp\n");
+		      return ERANGE;
+		    }
+		  mpd->state = 0;
+		} /* switch (state) */
+	    } /* if attribute_is_deleted() */
+	} /* message_get_attribute() */
+    } /* for */
+  mpd->func = mpd->id = NULL;
+  mpd->state = 0;
 
   return 0;
 }
@@ -1180,8 +1284,6 @@ bio_readline (bio_t bio)
       memmove (bio->buffer, bio->nl + 1, bio->ptr - bio->nl);
       bio->ptr = bio->buffer + (bio->ptr - bio->nl) - 1;
       bio->nl = memchr (bio->buffer, '\n', bio->ptr - bio->buffer + 1);
-      //if (bio->nl == NULL)
-      //bio->nl = bio->buffer;
     }
   else
     bio->nl = bio->ptr = bio->buffer;
@@ -1198,10 +1300,11 @@ bio_readline (bio_t bio)
 	  bio->nl = memchr (bio->buffer, '\n', len);
 	  if (bio->nl == NULL)
 	    {
-	      if (len >= bio->maxlen)
+	      if (len >= (bio->maxlen - 1))
 		{
-		  char *tbuf = realloc (bio->buffer,
-				    (2*(bio->maxlen) + 1)*(sizeof(char)));
+		  char *tbuf;
+		  tbuf = realloc (bio->buffer,
+				  (2*(bio->maxlen) + 2)*(sizeof(char)));
 		  if (tbuf == NULL)
 		    return ENOMEM;
 		  bio->buffer = tbuf;

@@ -312,9 +312,10 @@ mailbox_unix_destroy (mailbox_t *pmbox)
 	      if (mum == NULL)
 		continue;
 	      /* Destroy the attach messages */
-	      message_destroy (&(mum->message), mum);
 	      attribute_destroy (&(mum->old_attr));
-	      attribute_destroy (&(mum->new_attr));
+              message_destroy (&(mum->message), mum);
+	      /* new_attr free by message_destroy() */
+	      /* attribute_destroy (&(mum->new_attr)); */
 	      free (mum);
 	    }
 	  free (mud->umessages);
@@ -474,6 +475,11 @@ mailbox_unix_open (mailbox_t mbox, int flags)
     }
     funlockfile (mud->file);
     mud->flags = flags;
+
+    /* give an appriate way to lock */
+    if (mbox->locker == NULL)
+      locker_create (&(mbox->locker), mbox->name,
+		     strlen (mbox->name), MU_LOCKER_PID | MU_LOCKER_FCNTL);
   }
   mailbox_unix_iunlock (mbox);
   return 0;
@@ -547,7 +553,7 @@ mailbox_unix_num_deleted (mailbox_t mbox, size_t *pnum)
   two copies.
 */
 static FILE *
-mailbox_unix_tmpfile (mailbox_t mbox, char *tmpmbox)
+mailbox_unix_tmpfile (mailbox_t mbox, char **pbox)
 {
   mailbox_unix_data_t mud = (mailbox_unix_data_t)mbox->data;
   int fd;
@@ -557,34 +563,34 @@ mailbox_unix_tmpfile (mailbox_t mbox, char *tmpmbox)
 # define P_tmpdir "/tmp"
 #endif
 
-  tmpmbox = calloc (1, strlen (P_tmpdir) + strlen ("MBOX_") +
+  *pbox = calloc (1, strlen (P_tmpdir) + strlen ("MBOX_") +
 		   strlen (mud->basename) + 1);
-  if (tmpmbox == NULL)
+  if (*pbox == NULL)
     return NULL;
-  sprintf (tmpmbox, "%s/%s%s", P_tmpdir, "MBOX_", mud->basename);
+  sprintf (*pbox, "%s/%s%s", P_tmpdir, "MBOX_", mud->basename);
 
   /* FIXME:  I don't think this is the righ approach
    * Creating an anonymous file would be better ?
    * no trace left behind.
    */
   /* Create the file.  It must not exist.  If it does exist, fail. */
-  fd = open(tmpmbox, O_RDWR|O_CREAT|O_EXCL, 0600);
+  fd = open(*pbox, O_RDWR|O_CREAT|O_EXCL, 0600);
   if (fd < 0)
     {
-      fprintf(stderr,"Can't create %s\n", tmpmbox);
-      fprintf(stderr,"delete file <%s>, Please\n", tmpmbox);
+      fprintf(stderr,"Can't create %s\n", *pbox);
+      fprintf(stderr,"delete file <%s>, Please\n", *pbox);
       return NULL;
     }
   fp = fdopen(fd, "w+");
   if (fp == 0)
     {
       close(fd);
-      free (tmpmbox);
-      tmpmbox = NULL;
+      free (*pbox);
+      *pbox = NULL;
     }
 
   /* really I should just remove the file here */
-  /* remove(tmpmbox); */
+  /* remove(*pbox); */
   return fp;
 }
 
@@ -612,7 +618,7 @@ mailbox_unix_expunge (mailbox_t mbox)
   if (mud->messages_count == 0)
     return 0;
 
-  tempfile = mailbox_unix_tmpfile (mbox, tmpmbox);
+  tempfile = mailbox_unix_tmpfile (mbox, &tmpmbox);
   if (tempfile == NULL)
     return errno;
 
@@ -644,7 +650,8 @@ mailbox_unix_expunge (mailbox_t mbox)
 
   /*
    * We can not be NONBLOCKING here.
-   * It would irresponsable.
+   * It would irresponsable. Then again, it suppose
+   * to be a __LOCAL__ file system
    */
   if ((oflags = fcntl (fileno (mud->file), F_GETFL, 0)) < 0)
     {
@@ -672,7 +679,8 @@ mailbox_unix_expunge (mailbox_t mbox)
   for (j = 0; j < mud->messages_count; j++)
     {
       mum = mud->umessages[j];
-      if (! attribute_is_equal (mum->old_attr, mum->new_attr))
+      if (mum->new_attr &&
+	  ! attribute_is_equal (mum->old_attr, mum->new_attr))
 	break;
     }
 
@@ -681,7 +689,7 @@ mailbox_unix_expunge (mailbox_t mbox)
     {
       /* nothing change, bail out */
       status = 0;
-      goto bailout;
+      goto bailout_ok;
     }
 
   /* set the marker position */
@@ -732,31 +740,12 @@ mailbox_unix_expunge (mailbox_t mbox)
 	    }
 	  /* put the new attributes */
 	  {
-	    attribute_t attr = mum->new_attr;
-	    fputs ("Status: ", tempfile);
-	    total += 8;
-	    if (attribute_is_seen (attr))
-	      {
-		fputc ('R', tempfile);
-		total++;
-	      }
-	    if (attribute_is_answered (attr))
-	      {
-		fputc ('A', tempfile);
-		total++;
-	      }
-	    if (attribute_is_flagged (attr))
-	      {
-		fputc ('F', tempfile);
-		total++;
-	      }
-	    if (attribute_is_read (attr))
-	      {
-		fputc ('O', tempfile);
-		total++;
-	      }
-	    fputc ('\n', tempfile);
-	    total++;
+	    char abuf[64];
+	    size_t na = 0;
+	    abuf[0] = '\0';
+	    attribute_to_string (mum->new_attr, abuf, sizeof(abuf), &na);
+	    fputs (abuf, tempfile);
+	    total += na;
 	  }
 	  /* skip the status field */
 	  if (fseek (mud->file, current, SEEK_SET) == -1)
@@ -843,7 +832,7 @@ mailbox_unix_expunge (mailbox_t mbox)
       }
   }
   /* Seek and rewrite it */
-  if (total <= 0 || fseek (mud->file, marker, SEEK_SET) < 0)
+  if (total < 0 || fseek (mud->file, marker, SEEK_SET) < 0)
     {
       status = errno;
       goto bailout;
@@ -877,10 +866,11 @@ mailbox_unix_expunge (mailbox_t mbox)
 
   /* FIXME: update the num of all the messages */
 
+ bailout_ok:
   /* Don't remove the tmp mbox in case of errors */
   remove (tmpmbox);
 
-bailout:
+ bailout:
   free (tmpmbox);
   /* Release the locks */
   if (oflags > 0)
@@ -1181,11 +1171,14 @@ mailbox_unix_get_message (mailbox_t mbox, size_t msgno, message_t *pmsg)
     }
   stream_set_read (stream, mailbox_unix_readstream, mum);
   stream_set_fd (stream, mailbox_unix_getfd, mum);
+  stream_set_flags (stream, mud->flags, mum);
   body_set_stream (body, stream, mum);
   body_set_size (body, mailbox_unix_body_size, mum);
   body_set_lines (body, mailbox_unix_body_lines, mum);
 
   /* set the attribute */
+  attribute_create (&(mum->new_attr));
+  mum->new_attr->flag = mum->old_attr->flag;
   status = message_set_attribute (msg, mum->new_attr, mum);
   if (status != 0)
     {
@@ -1247,9 +1240,6 @@ mailbox_unix_append_message (mailbox_t mbox, message_t msg)
       }
 
     /* header */
-    if (st.st_size != 0)
-      fputc ('\n', mud->file);
-
     message_get_header (msg, &hdr);
     /* generate a "From " separator */
     {
@@ -1289,6 +1279,7 @@ mailbox_unix_append_message (mailbox_t mbox, message_t msg)
       fwrite (buffer, sizeof (*buffer), nread, mud->file);
       off += nread;
     } while (nread > 0);
+    fputc ('\n', mud->file);
   }
   fflush(mud->file);
   funlockfile (mud->file);

@@ -267,16 +267,15 @@ do \
     } \
 } while (0)
 
+//    size_t num = 2 * ((mud)->messages_count) + 10;
 /* allocate slots for the new messages */
 #define ALLOCATE_MSGS(mbox,mud,file) \
 do \
 { \
-  if (mud->umessages_count == 0 || \
-      ((mud)->umessages_count + 1) <= (mud)->messages_count) \
+  if ((mud)->messages_count >= (mud)->umessages_count) \
   { \
     mailbox_unix_message_t *m; \
-    size_t i; \
-    size_t num = 2 * ((mud)->messages_count) + 10; \
+    size_t num = ((mud)->umessages_count) + 1; \
     m = realloc ((mud)->umessages, num * sizeof (*m)); \
     if (m == NULL) \
       { \
@@ -286,24 +285,16 @@ do \
         return ENOMEM; \
       } \
     (mud)->umessages = m; \
-    if ((mud)->umessages_count) \
-      i = (mud)->umessages_count + 1; \
-    else \
-      i =  0; \
-    for (; i < num; i++) \
+    (mud)->umessages[num - 1] = calloc (1, sizeof (*(mum))); \
+    if ((mud)->umessages[num - 1] == NULL) \
       { \
-        (mud)->umessages[i] = calloc (1, sizeof (*(mum))); \
-        if ((mud)->umessages[i] == NULL) \
-          { \
-            fclose (file); \
-            mailbox_unix_iunlock (mbox); \
-            mailbox_unix_unlock (mbox); \
-            return ENOMEM; \
-          } \
-        ATTRIBUTE_CREATE (((mud)->umessages[i])->old_attr, mbox); \
-        ATTRIBUTE_CREATE (((mud)->umessages[i])->new_attr, mbox); \
+        fclose (file); \
+        mailbox_unix_iunlock (mbox); \
+        mailbox_unix_unlock (mbox); \
+        return ENOMEM; \
       } \
-    (mud)->umessages_count = num - 1; \
+    ATTRIBUTE_CREATE (((mud)->umessages[num - 1])->old_attr, mbox); \
+    (mud)->umessages_count = num; \
   } \
 } while (0)
 
@@ -433,7 +424,6 @@ mailbox_unix_scan (mailbox_t mbox, size_t msgno, size_t *pcount)
 	      ATTRIBUTE_SET(buf, mum, 'o', 'O', MU_ATTRIBUTE_SEEN);
 	      ATTRIBUTE_SET(buf, mum, 'a', 'A', MU_ATTRIBUTE_ANSWERED);
 	      ATTRIBUTE_SET(buf, mum, 'd', 'D', MU_ATTRIBUTE_DELETED);
-	      mum->new_attr->flag = mum->old_attr->flag;
 	    }
 	}
 

@@ -39,22 +39,12 @@ int
 message_create (message_t *pmsg, void *owner)
 {
   message_t msg;
-  stream_t stream;
-  int status;
 
   if (pmsg == NULL)
     return EINVAL;
   msg = calloc (1, sizeof (*msg));
   if (msg == NULL)
     return ENOMEM;
-  status = stream_create (&stream, MU_STREAM_RDWR, msg);
-  if (status != 0)
-    {
-      free (msg);
-      return status;
-    }
-  stream_set_read (stream, message_read, msg);
-  stream_set_write (stream, message_write, msg);
   msg->owner = owner;
   *pmsg = msg;
   return 0;
@@ -66,48 +56,32 @@ message_destroy (message_t *pmsg, void *owner)
   if (pmsg && *pmsg)
     {
       message_t msg = *pmsg;
-      int destroy = 0;
 
-      /* always decremente */
-      msg->ref_count--;
-
-      if (msg->owner && msg->owner == owner)
-	destroy = 1;
-
-      if (msg->owner == NULL && msg->ref_count <= 0)
+      if (msg->owner == owner)
 	{
-	  destroy = 1;
-	  owner = msg;
-	}
-
-      if (destroy)
-	{
-	  attribute_t attribute = msg->attribute;
-	  stream_t stream = msg->stream;
-	  header_t header = msg->header;
-	  body_t body = msg->body;
-
 	  /* notify the listeners */
-	  message_notification (msg, MU_EVT_MSG_DESTROY);
+	  if (msg->event_num)
+	    {
+	      message_notification (msg, MU_EVT_MSG_DESTROY);
+	      free (msg->event);
+	    }
 	  /* header */
-	  header_destroy (&header, owner);
+	  header_destroy (&(msg->header), owner);
 	  /* attribute */
-	  attribute_destroy (&attribute);
+	  attribute_destroy (&(msg->attribute));
 	  /* stream */
-	  stream_destroy (&stream, owner);
+	  stream_destroy (&(msg->stream), owner);
 
 	  /* if sometype of floating/temporary message */
-	  body_destroy (&body, owner);
+	  body_destroy (&(msg->body), owner);
 	  /* notifications are done */
-	  free (msg->event);
 
 	  /* check again for resurrection before free()ing
 	   * the memory maybe it was clone, if yes we can not
 	   * free the pointer.
 	   *
 	   */
-	  if (msg->ref_count <= 0)
-	    free (msg);
+	  free (msg);
 	}
       /* loose the link */
       *pmsg = NULL;
@@ -347,7 +321,7 @@ message_get_attribute (message_t msg, attribute_t *pattribute)
 {
   if (msg == NULL || pattribute == NULL)
    return EINVAL;
-  if (msg->attribute == NULL && msg->owner == NULL)
+  if (msg->attribute == NULL)
     {
       attribute_t attribute;
       int status = attribute_create (&attribute);