Rewrote to use channel_io.c.

Rev: src/channel_forward.c:1.14.2.1
Rev: src/channel_forward.h:1.8.2.1

src/channel_forward.c

@@ -34,6 +34,7 @@
 #include "channel_forward.h"
 
 #include "io.h"
+#include "lsh_string.h"
 #include "ssh.h"
 #include "werror.h"
 #include "xalloc.h"
@@ -42,33 +43,44 @@
 #include "channel_forward.h.x"
 #undef GABA_DEFINE
 
-/* NOTE: Adds the socket to the channel's resource list */
+/* We don't use channel_read_state_close and channel_write_state_close. */
+static void
+do_kill_channel_forward(struct resource *s)
+{  
+  CAST(channel_forward, self, s);
+  if (self->super.super.alive)
+    {
+      trace("do_kill_channel_forward\n");
+      
+      self->super.super.alive = 0;
+      io_close_fd(self->read.fd);
+    }
+}
+
+#define FORWARD_READ_BUFFER_SIZE 0x4000
+
 void
 init_channel_forward(struct channel_forward *self,
-		     struct lsh_fd *socket, uint32_t initial_window)
+		     int socket, uint32_t initial_window)
 {
-  assert(socket);
-  
-  init_channel(&self->super);
+  init_channel(&self->super, do_kill_channel_forward);
 
   /* The rest of the callbacks are not set up until
    * channel_forward_start_io. */
 
-  /* NOTE: We don't need a close handler, as the channel's resource
-   * list is taken care of automatically. */
-  
   self->super.rec_window_size = initial_window;
 
   /* FIXME: Make maximum packet size configurable. */
   self->super.rec_max_packet = SSH_MAX_PACKET;
-  
-  self->socket = socket;
 
-  remember_resource(self->super.resources, &socket->super);
+  init_channel_read_state(&self->read, socket, FORWARD_READ_BUFFER_SIZE);
+  init_channel_write_state(&self->write, socket, initial_window);
+
+  io_register_fd(socket, "forwarded socket");
 }
 
 struct channel_forward *
-make_channel_forward(struct lsh_fd *socket, uint32_t initial_window)
+make_channel_forward(int socket, uint32_t initial_window)
 {
   NEW(channel_forward, self);
   init_channel_forward(self, socket, initial_window);
@@ -77,32 +89,73 @@ make_channel_forward(struct lsh_fd *socket, uint32_t initial_window)
 }
 
 static void
-do_channel_forward_receive(struct ssh_channel *c,
-			   int type, struct lsh_string *data)
+channel_forward_shutdown(struct channel_forward *self)
 {
-  CAST_SUBTYPE(channel_forward, closure, c);
+  if (shutdown (self->write.fd, SHUT_WR) < 0)
+    werror("close_fd_write, shutdown failed, %e\n", errno);
+
+  assert(self->super.sinks);
+  self->super.sinks--;
+  channel_maybe_close(&self->super);  
+}
+
+static void *
+oop_write_socket(oop_source *source UNUSED,
+		 int fd, oop_event event, void *state)
+{
+  CAST(channel_forward, self, (struct lsh_object *) state);
   
-  switch (type)
+  assert(event == OOP_WRITE);
+  assert(fd == self->write.fd);
+
+  if (channel_io_flush(&self->super, &self->write) == CHANNEL_IO_EOF)
+    channel_forward_shutdown(self);
+
+  return OOP_CONTINUE;
+}
+
+static void
+do_channel_forward_receive(struct ssh_channel *s, int type,
+			   uint32_t length, const uint8_t *data)
+{
+  CAST_SUBTYPE(channel_forward, self, s);
+
+  if (type != CHANNEL_DATA)
+    werror("Ignoring unexpected extended data on forwarded channel.\n");
+  else
     {
-    case CHANNEL_DATA:
-      A_WRITE(&closure->socket->write_buffer->super, data);
-      break;
-    case CHANNEL_STDERR_DATA:
-      werror("Ignoring unexpected stderr data.\n");
-      lsh_string_free(data);
-      break;
-    default:
-      fatal("Internal error. do_channel_forward_receive");
+      if (channel_io_write(&self->super, &self->write, oop_write_socket,
+			   length, data) == CHANNEL_IO_EOF)
+	channel_forward_shutdown(self);
     }
 }
 
+static void *
+oop_read_socket(oop_source *source UNUSED,
+		int fd, oop_event event, void *state)
+{
+  CAST(channel_forward, self, (struct lsh_object *) state);
+  uint32_t done;
+  
+  assert(fd == self->read.fd);
+  assert(event == OOP_READ);
+
+  if (channel_io_read(&self->super, &self->read, &done) == CHANNEL_IO_OK
+      && done > 0)
+    channel_transmit_data(&self->super,
+			  done, lsh_string_data(self->read.buffer));
+
+  return OOP_CONTINUE;
+}
+
+/* We may send more data */
 static void
 do_channel_forward_send_adjust(struct ssh_channel *s,
 			       uint32_t i UNUSED)
 {
-  CAST_SUBTYPE(channel_forward, self, s);
-  
-  lsh_oop_register_read_fd(self->socket);
+  CAST(channel_forward, self, s);
+
+  channel_io_start_read(&self->super, &self->read, oop_read_socket);
 }
 
 static void
@@ -110,36 +163,30 @@ do_channel_forward_eof(struct ssh_channel *s)
 {
   CAST_SUBTYPE(channel_forward, self, s);
 
-  /* We won't write any more. io.c should make sure that shutdown is called
-   * once the write_buffer is empty. */
-  close_fd_write(self->socket);
+  if (!self->write.state->length)
+    channel_forward_shutdown(self);
 }
 
-
 /* NOTE: Because this function is called by
  * do_open_forwarded_tcpip_continuation, the same restrictions apply.
  * I.e we can not assume that the channel is completely initialized
  * (channel_open_continuation has not yet done its work), and we can't
  * send any packets. */
 void
-channel_forward_start_io(struct channel_forward *channel)
+channel_forward_start_io(struct channel_forward *self)
 {
-  channel->super.receive = do_channel_forward_receive;
-  channel->super.send_adjust = do_channel_forward_send_adjust;
-  channel->super.eof = do_channel_forward_eof;
-  
-  /* Install callbacks on the local socket.
-   * NOTE: We don't install any channel_io_exception_handler. */
-  io_read_write(channel->socket,
-		make_channel_read_data(&channel->super),
-		/* FIXME: Make this configurable */
-		SSH_MAX_PACKET * 10, /* self->block_size, */
-		make_channel_read_close_callback(&channel->super));
+  self->super.receive = do_channel_forward_receive;
+  self->super.send_adjust = do_channel_forward_send_adjust;
+  self->super.eof = do_channel_forward_eof;
+
+  self->super.sources++;
+  self->super.sinks ++;
   
-  /* Flow control */
-  channel->socket->write_buffer->report = &channel->super.super;
+  if (self->super.send_window_size)
+    channel_io_start_read(&self->super, &self->read, oop_read_socket);
 }
 
+#if 0
 /* Like above, but doesn't install a new write buffer. Used by the socks server. */
 void
 channel_forward_start_io_read(struct channel_forward *channel)
@@ -156,13 +203,14 @@ channel_forward_start_io_read(struct channel_forward *channel)
   /* Flow control */
   channel->socket->write_buffer->report = &channel->super.super;
 }
+#endif
 
 /* Used by the party requesting tcp forwarding, i.e. when a socket is
  * already open, and we have asked the other end to forward it. Takes
  * a channel as argument, and connects it to the socket. Returns the
  * channel. */
 
-DEFINE_COMMAND(start_io_command)
+DEFINE_COMMAND(forward_start_io_command)
      (struct command *s UNUSED,
       struct lsh_object *x,
       struct command_continuation *c,

src/channel_forward.h

@@ -27,6 +27,7 @@
 #define LSH_CHANNEL_FORWARD_H_INCLUDED
 
 #include "channel.h"
+#include "channel_io.h"
 
 #define GABA_DECLARE
 #include "channel_forward.h.x"
@@ -37,15 +38,17 @@
      (name channel_forward)
      (super ssh_channel)
      (vars
-       (socket object lsh_fd)))
+       ; We have to be careful, since both structs use the same fd
+       (read struct channel_read_state)
+       (write struct channel_write_state)))
 */
 
 void
 init_channel_forward(struct channel_forward *self,
-		     struct lsh_fd *socket, uint32_t initial_window);
+		     int fd, uint32_t initial_window);
 
 struct channel_forward *
-make_channel_forward(struct lsh_fd *socket, uint32_t initial_window);
+make_channel_forward(int fd, uint32_t initial_window);
 
 void
 channel_forward_start_io(struct channel_forward *channel_forward);
@@ -53,8 +56,8 @@ channel_forward_start_io(struct channel_forward *channel_forward);
 void
 channel_forward_start_io_read(struct channel_forward *channel);
 
-extern struct command start_io_command;
-#define START_IO (&start_io_command.super)
+extern struct command forward_start_io_command;
+#define FORWARD_START_IO (&forward_start_io_command.super)
 
 extern struct catch_report_collect catch_channel_open;
 #define CATCH_CHANNEL_OPEN (&catch_channel_open.super.super)