* mailbox/socket_stream.c: New file.

* include/mailutils/stream.h (mu_socket_stream_create)
(mu_stream_shutdown, mu_stream_set_shutdown): New functions.
* libproto/include/stream0.h (struct _mu_stream): New member
`_shutdown'.
* mailbox/file_stream.c (mu_stream_create): Bugfix.
* mailbox/stream.c (mu_stream_shutdown)
(mu_stream_set_shutdown): New functions.
* mailbox/tcp.c (_tcp_shutdown): New function.
(_tcp_stream_init): New function.
(mu_tcp_stream_create_with_source_ip): Register _tcp_shutdown.

@@ -56,6 +56,8 @@ extern int mu_tcp_stream_create_with_source_host (mu_stream_t *stream,
 						  const char *host, int port,
 						  const char *host, int port,
 						  const char *source_host,
 						  const char *source_host,
 						  int flags);
 						  int flags);
+extern int mu_socket_stream_create (mu_stream_t *stream, const char *filename,
+				    int flags);
   
   
 extern int mu_mapfile_stream_create (mu_stream_t *stream, const char* filename,
 extern int mu_mapfile_stream_create (mu_stream_t *stream, const char* filename,
 				     int flags);
 				     int flags);
@@ -91,6 +93,7 @@ extern int mu_stream_write       (mu_stream_t, const char *, size_t, mu_off_t,
 				  size_t *);
 				  size_t *);
 extern int mu_stream_setbufsiz   (mu_stream_t stream, size_t size);
 extern int mu_stream_setbufsiz   (mu_stream_t stream, size_t size);
 extern int mu_stream_flush       (mu_stream_t);
 extern int mu_stream_flush       (mu_stream_t);
+extern int mu_stream_shutdown    (mu_stream_t stream, int how);
 
 
 extern int mu_stream_vprintf     (mu_stream_t os, mu_off_t *poff, 
 extern int mu_stream_vprintf     (mu_stream_t os, mu_off_t *poff, 
                                   const char *fmt, va_list ap);
                                   const char *fmt, va_list ap);
@@ -166,6 +169,9 @@ extern int mu_stream_set_strerror (mu_stream_t stream,
 
 
 extern int mu_stream_set_wait (mu_stream_t stream,
 extern int mu_stream_set_wait (mu_stream_t stream,
       int (*wait) (mu_stream_t, int *, struct timeval *), void *owner);
       int (*wait) (mu_stream_t, int *, struct timeval *), void *owner);
+
+extern int mu_stream_set_shutdown (mu_stream_t stream,
+      int (*_shutdown) (mu_stream_t, int how), void *owner);
   
   
 extern int mu_stream_sequential_read (mu_stream_t stream,
 extern int mu_stream_sequential_read (mu_stream_t stream,
       char *buf, size_t size, size_t *nbytes);
       char *buf, size_t size, size_t *nbytes);

@@ -65,6 +65,7 @@ struct _mu_stream
   int (*_setbufsiz)(mu_stream_t, size_t);
   int (*_setbufsiz)(mu_stream_t, size_t);
   int (*_strerror) (mu_stream_t, const char **);
   int (*_strerror) (mu_stream_t, const char **);
   int (*_wait)     (mu_stream_t, int *pflags, struct timeval *tvp);
   int (*_wait)     (mu_stream_t, int *pflags, struct timeval *tvp);
+  int (*_shutdown) (mu_stream_t, int how);
 };
 };
 
 
 #ifdef __cplusplus
 #ifdef __cplusplus

@@ -546,8 +546,8 @@ mu_file_stream_create (mu_stream_t *stream, const char* filename, int flags)
   ret = mu_stream_create (stream, flags|MU_STREAM_NO_CHECK, fs);
   ret = mu_stream_create (stream, flags|MU_STREAM_NO_CHECK, fs);
   if (ret != 0)
   if (ret != 0)
     {
     {
-      free (fs);
       free (fs->filename);
       free (fs->filename);
+      free (fs);
       return ret;
       return ret;
     }
     }
 
 

+/* GNU Mailutils -- a suite of utilities for electronic mail
+   Copyright (C) 1999, 2000, 2001, 2002, 2004, 
+   2005, 2006, 2007 Free Software Foundation, Inc.
+
+   This library is free software; you can redistribute it and/or
+   modify it under the terms of the GNU Lesser General Public
+   License as published by the Free Software Foundation; either
+   version 3 of the License, or (at your option) any later version.
+
+   This library 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
+   Lesser General Public License for more details.
+
+   You should have received a copy of the GNU Lesser General
+   Public License along with this library; if not, write to the
+   Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+   Boston, MA 02110-1301 USA */
+
+#ifdef HAVE_CONFIG_H
+# include <config.h>
+#endif
+
+#include <unistd.h>
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <string.h>
+#include <signal.h>
+#include <errno.h>
+
+#include <mailutils/stream.h>
+#include <mailutils/error.h>
+#include <mailutils/errno.h>
+
+struct _socket_stream
+{
+  mu_stream_t fstream;
+  char *filename;
+  int ec; /* Last error code if fstream == NULL */
+};
+
+static void
+_s_destroy (mu_stream_t stream)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+
+  if (s->filename)
+    free (s->filename);
+  mu_stream_destroy (&s->fstream, mu_stream_get_owner (s->fstream));
+  free (s);
+}
+
+static int
+_s_read (mu_stream_t stream, char *optr, size_t osize,
+	 mu_off_t offset, size_t *nbytes)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  return mu_stream_read (s->fstream, optr, osize, offset, nbytes);
+}
+
+static int
+_s_readline (mu_stream_t stream, char *optr, size_t osize,
+	     mu_off_t offset, size_t *nbytes)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  return mu_stream_readline (s->fstream, optr, osize, offset, nbytes);
+}
+
+static int
+_s_write (mu_stream_t stream, const char *iptr, size_t isize,
+	  mu_off_t offset, size_t *nbytes)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  return mu_stream_write (s->fstream, iptr, isize, offset, nbytes);
+}
+
+static int
+_s_open (mu_stream_t stream)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  int fd, rc;
+  FILE *fp;
+  struct sockaddr_un addr;
+  char *fstr;
+  int flags;
+  
+  if (!s)
+    return EINVAL;
+  
+  fd = socket (PF_UNIX, SOCK_STREAM, 0);
+  if (fd < 0)
+    return errno;
+  
+  memset (&addr, 0, sizeof addr);
+  addr.sun_family = AF_UNIX;
+  strncpy (addr.sun_path, s->filename, sizeof addr.sun_path - 1);
+  addr.sun_path[sizeof addr.sun_path - 1] = 0;
+  if (connect (fd, (struct sockaddr *) &addr, sizeof(addr)))
+    {
+      close (fd);
+      return errno;
+    }
+
+  mu_stream_get_flags(stream, &flags);
+  if (flags & MU_STREAM_WRITE)
+    fstr = "w";
+  else if (flags & MU_STREAM_RDWR)
+    fstr = "w+";
+  else /* default, also if flags & MU_STREAM_READ */
+    fstr = "r";
+  
+  fp = fdopen (fd, fstr);
+  if (!fp)
+    {
+      close (fd);
+      return errno;
+    }
+      
+  rc = mu_stdio_stream_create (&s->fstream, fp, flags);
+  if (rc)
+    {
+      fclose (fp);
+      return rc;
+    }
+  
+  rc = mu_stream_open (s->fstream);
+  if (rc)
+    {
+      mu_stream_destroy (&s->fstream, mu_stream_get_owner (s->fstream));
+      fclose (fp);
+    }
+  return rc;
+}
+
+static int
+_s_close (mu_stream_t stream)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  return mu_stream_close (s->fstream);
+}
+
+static int
+_s_flush (mu_stream_t stream)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  return mu_stream_flush (s->fstream);
+}
+
+int
+_s_wait (mu_stream_t stream, int *pflags, struct timeval *tvp)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  return mu_stream_wait (s->fstream, pflags, tvp);
+}
+
+int
+_s_strerror (mu_stream_t stream, const char **pstr)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  return mu_stream_strerror (s->fstream, pstr);
+}
+
+static int
+_s_get_transport2 (mu_stream_t stream,
+		   mu_transport_t *pin, mu_transport_t *pout)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  return mu_stream_get_transport2 (s->fstream, pin, pout);
+}
+
+int
+_s_shutdown (mu_stream_t stream, int how)
+{
+  struct _socket_stream *s = mu_stream_get_owner (stream);
+  int flag;
+  mu_transport_t trans;
+
+  if (s->fstream == NULL)
+    return EINVAL;
+
+  mu_stream_get_transport(s->fstream, &trans);
+  switch (how)
+    {
+    case MU_STREAM_READ:
+      flag = SHUT_RD;
+      break;
+      
+    case MU_STREAM_WRITE:
+      flag = SHUT_WR;
+    }
+
+  if (shutdown ((int) trans, flag))
+    return errno;
+  return 0;
+}
+
+int
+mu_socket_stream_create (mu_stream_t *stream, const char *filename, int flags)
+{
+  struct _socket_stream *s;
+  int rc;
+
+  if (stream == NULL)
+    return MU_ERR_OUT_PTR_NULL;
+
+  s = calloc (1, sizeof (struct _socket_stream));
+  if (s == NULL)
+    return ENOMEM;
+
+  if ((s->filename = strdup (filename)) == NULL)
+    {
+      free (s);
+      return ENOMEM;
+    }
+
+  rc = mu_stream_create (stream, flags | MU_STREAM_NO_CHECK, s);
+  if (rc)
+    {
+      free (s);
+      free (s->filename);
+      return rc;
+    }
+
+  mu_stream_set_open (*stream, _s_open, s);
+  mu_stream_set_close (*stream, _s_close, s);
+  mu_stream_set_get_transport2 (*stream, _s_get_transport2, s);
+  mu_stream_set_read (*stream, _s_read, s);
+  mu_stream_set_readline (*stream, _s_readline, s);
+  mu_stream_set_write (*stream, _s_write, s);
+  mu_stream_set_flush (*stream, _s_flush, s);
+  mu_stream_set_destroy (*stream, _s_destroy, s);
+  mu_stream_set_strerror (*stream, _s_strerror, s);
+  mu_stream_set_wait (*stream, _s_wait, s);
+  mu_stream_set_shutdown (*stream, _s_shutdown, s);
+  
+  return 0;
+}
+
+
+
+

@@ -669,6 +669,25 @@ mu_stream_get_state (mu_stream_t stream, int *pstate)
 }
 }
 
 
 int
 int
+mu_stream_shutdown (mu_stream_t stream, int how)
+{
+  if (stream == NULL)
+    return EINVAL;
+  if (!stream->_shutdown)
+    return ENOSYS;
+  switch (how)
+    {
+    case MU_STREAM_READ:
+    case MU_STREAM_WRITE:
+      break;
+
+    default:
+      return EINVAL;
+    }
+  return stream->_shutdown (stream, how);
+}
+
+int
 mu_stream_set_destroy (mu_stream_t stream,
 mu_stream_set_destroy (mu_stream_t stream,
 		       void (*_destroy) (mu_stream_t), void *owner)
 		       void (*_destroy) (mu_stream_t), void *owner)
 {
 {
@@ -849,6 +868,20 @@ mu_stream_set_wait (mu_stream_t stream,
 }
 }
 
 
 int
 int
+mu_stream_set_shutdown (mu_stream_t stream,
+			int (*_shutdown) (mu_stream_t, int how), void *owner)
+{
+  if (stream == NULL)
+    return EINVAL;
+  if (owner == stream->owner)
+    {
+      stream->_shutdown = _shutdown;
+      return 0;
+    }
+  return EACCES;
+}
+
+int
 mu_stream_sequential_read (mu_stream_t stream, char *buf, size_t size,
 mu_stream_sequential_read (mu_stream_t stream, char *buf, size_t size,
 			   size_t *nbytes)
 			   size_t *nbytes)
 {
 {

@@ -266,6 +266,42 @@ _tcp_wait (mu_stream_t stream, int *pflags, struct timeval *tvp)
 }
 }
 
 
 int
 int
+_tcp_shutdown (mu_stream_t stream, int how)
+{
+  struct _tcp_instance *tcp = mu_stream_get_owner (stream);
+  int flag;
+  if (tcp->fd == -1)
+    return EINVAL;
+
+  switch (how)
+    {
+    case MU_STREAM_READ:
+      flag = SHUT_RD;
+      break;
+      
+    case MU_STREAM_WRITE:
+      flag = SHUT_WR;
+    }
+
+  if (shutdown (tcp->fd, flag))
+    return errno;
+  return 0;
+}
+
+static void
+_tcp_stream_init (mu_stream_t stream, struct _tcp_instance *tcp)
+{
+  mu_stream_set_open (stream, _tcp_open, tcp);
+  mu_stream_set_close (stream, _tcp_close, tcp);
+  mu_stream_set_read (stream, _tcp_read, tcp);
+  mu_stream_set_write (stream, _tcp_write, tcp);
+  mu_stream_set_get_transport2 (stream, _tcp_get_transport2, tcp);
+  mu_stream_set_destroy (stream, _tcp_destroy, tcp);
+  mu_stream_set_wait (stream, _tcp_wait, tcp);
+  mu_stream_set_shutdown (stream, _tcp_shutdown, tcp);
+}
+
+int
 mu_tcp_stream_create_with_source_ip (mu_stream_t *stream,
 mu_tcp_stream_create_with_source_ip (mu_stream_t *stream,
 				     const char *host, int port,
 				     const char *host, int port,
 				     unsigned long source_ip,
 				     unsigned long source_ip,
@@ -301,14 +337,7 @@ mu_tcp_stream_create_with_source_ip (mu_stream_t *stream,
     return ret;
     return ret;
   }
   }
 
 
-  mu_stream_set_open (*stream, _tcp_open, tcp);
-  mu_stream_set_close (*stream, _tcp_close, tcp);
-  mu_stream_set_read (*stream, _tcp_read, tcp);
-  mu_stream_set_write (*stream, _tcp_write, tcp);
-  mu_stream_set_get_transport2 (*stream, _tcp_get_transport2, tcp);
-  mu_stream_set_destroy (*stream, _tcp_destroy, tcp);
-  mu_stream_set_wait (*stream, _tcp_wait, tcp);
-
+  _tcp_stream_init (*stream, tcp);
   return 0;
   return 0;
 }
 }