Added SPDY/3 flow control.

examples/SpdyServer.cc

@@ -173,7 +173,7 @@ SpdyEventHandler::SpdyEventHandler(const Config* config,
   : EventHandler(config),
     fd_(fd), ssl_(ssl), session_id_(session_id), want_write_(false)
 {
-  spdylay_session_server_new(&session_, callbacks, this);
+  spdylay_session_server_new(&session_, SPDYLAY_PROTO_SPDY2, callbacks, this);
 }
     
 SpdyEventHandler::~SpdyEventHandler()

examples/spdylay_ssl.cc

@@ -51,7 +51,7 @@ bool ssl_debug = false;
 Spdylay::Spdylay(int fd, SSL *ssl, const spdylay_session_callbacks *callbacks)
   : fd_(fd), ssl_(ssl), want_write_(false)
 {
-  spdylay_session_client_new(&session_, callbacks, this);
+  spdylay_session_client_new(&session_, SPDYLAY_PROTO_SPDY2, callbacks, this);
 }
 
 Spdylay::~Spdylay()
@@ -286,7 +286,8 @@ const char *ctrl_names[] = {
   "NOOP",
   "PING",
   "GOAWAY",
-  "HEADERS"
+  "HEADERS",
+  "WINDOW_UPDATE"
 };
 } // namespace
 
@@ -367,6 +368,12 @@ void print_frame(spdylay_frame_type type, spdylay_frame *frame)
     printf("(stream_id=%d)\n", frame->headers.stream_id);
     print_nv(frame->headers.nv);
     break;
+  case SPDYLAY_WINDOW_UPDATE:
+    print_frame_attr_indent();
+    printf("(stream_id=%d, delta_window_size=%d)\n",
+           frame->window_update.stream_id,
+           frame->window_update.delta_window_size);
+    break;
   default:
     printf("\n");
     break;

lib/includes/spdylay/spdylay.h

@@ -127,8 +127,8 @@ typedef enum {
   SPDYLAY_FLOW_CONTROL_ERROR = 7
 } spdylay_status_code;
 
-#define SPDYLAY_SPDY2_LOWEST_PRI 3
-#define SPDYLAY_SPDY3_LOWEST_PRI 7
+#define SPDYLAY_SPDY2_PRI_LOWEST 3
+#define SPDYLAY_SPDY3_PRI_LOWEST 7
 
 typedef struct {
   uint16_t version;
@@ -141,8 +141,8 @@ typedef struct {
   spdylay_ctrl_hd hd;
   int32_t stream_id;
   int32_t assoc_stream_id;
-  /* 0 (Highest) to SPDYLAY_SPDY2_LOWEST_PRI or
-     SPDYLAY_SPDY3_LOWEST_PRI (loweset), depending on the protocol
+  /* 0 (Highest) to SPDYLAY_SPDY2_PRI_LOWEST or
+     SPDYLAY_SPDY3_PRI_LOWEST (loweset), depending on the protocol
      version. */
   uint8_t pri;
   /* Since SPDY/3 */
@@ -372,10 +372,11 @@ typedef struct {
 } spdylay_session_callbacks;
 
 /*
- * Initializes |*session_ptr| for client use. The all members of
- * |callbacks| are copied to |*session_ptr|. Therefore |*session_ptr|
- * does not store |callbacks|. |user_data| is an arbitrary user
- * supplied data, which will be passed to the callback functions.
+ * Initializes |*session_ptr| for client use, using the protocol
+ * version |version|. The all members of |callbacks| are copied to
+ * |*session_ptr|. Therefore |*session_ptr| does not store
+ * |callbacks|. |user_data| is an arbitrary user supplied data, which
+ * will be passed to the callback functions.
  *
  * This function returns 0 if it succeeds, or one of the following
  * negative error codes:
@@ -388,14 +389,16 @@ typedef struct {
  *     The version is not supported.
  */
 int spdylay_session_client_new(spdylay_session **session_ptr,
+                               uint16_t version,
                                const spdylay_session_callbacks *callbacks,
                                void *user_data);
 
 /*
- * Initializes |*session_ptr| for server use. The all members of
- * |callbacks| are copied to |*session_ptr|. Therefore |*session_ptr|
- * does not store |callbacks|. |user_data| is an arbitrary user
- * supplied data, which will be passed to the callback functions.
+ * Initializes |*session_ptr| for server use, using the protocol
+ * version |version|. The all members of |callbacks| are copied to
+ * |*session_ptr|. Therefore |*session_ptr| does not store
+ * |callbacks|. |user_data| is an arbitrary user supplied data, which
+ * will be passed to the callback functions.
  *
  * This function returns 0 if it succeeds, or one of the following
  * negative error codes:
@@ -408,6 +411,7 @@ int spdylay_session_client_new(spdylay_session **session_ptr,
  *     The version is not supported.
  */
 int spdylay_session_server_new(spdylay_session **session_ptr,
+                               uint16_t version,
                                const spdylay_session_callbacks *callbacks,
                                void *user_data);
 

lib/spdylay_session.c

@@ -73,10 +73,14 @@ static int spdylay_outbound_item_compar(const void *lhsx, const void *rhsx)
 }
 
 static int spdylay_session_new(spdylay_session **session_ptr,
+                               uint16_t version,
                                const spdylay_session_callbacks *callbacks,
                                void *user_data)
 {
   int r;
+  if(version != SPDYLAY_PROTO_SPDY2 && version != SPDYLAY_PROTO_SPDY3) {
+    return SPDYLAY_ERR_UNSUPPORTED_VERSION;
+  }
   *session_ptr = malloc(sizeof(spdylay_session));
   if(*session_ptr == NULL) {
     r = SPDYLAY_ERR_NOMEM;
@@ -84,11 +88,15 @@ static int spdylay_session_new(spdylay_session **session_ptr,
   }
   memset(*session_ptr, 0, sizeof(spdylay_session));
 
+  (*session_ptr)->version = version;
+
   /* next_stream_id, last_recv_stream_id and next_unique_id are
      initialized in either spdylay_session_client_new or
      spdylay_session_server_new */
 
-  (*session_ptr)->version = SPDYLAY_PROTO_SPDY2;
+  (*session_ptr)->flow_control =
+    (*session_ptr)->version == SPDYLAY_PROTO_SPDY3;
+
   (*session_ptr)->last_ping_unique_id = 0;
 
   (*session_ptr)->next_seq = 0;
@@ -137,6 +145,9 @@ static int spdylay_session_new(spdylay_session **session_ptr,
   (*session_ptr)->settings[SPDYLAY_SETTINGS_MAX_CONCURRENT_STREAMS] =
     SPDYLAY_CONCURRENT_STREAMS_MAX;
 
+  (*session_ptr)->settings[SPDYLAY_SETTINGS_INITIAL_WINDOW_SIZE] =
+    SPDYLAY_INITIAL_WINDOW_SIZE;
+
   (*session_ptr)->callbacks = *callbacks;
   (*session_ptr)->user_data = user_data;
 
@@ -173,11 +184,12 @@ static int spdylay_session_new(spdylay_session **session_ptr,
 }
 
 int spdylay_session_client_new(spdylay_session **session_ptr,
+                               uint16_t version,
                                const spdylay_session_callbacks *callbacks,
                                void *user_data)
 {
   int r;
-  r = spdylay_session_new(session_ptr, callbacks, user_data);
+  r = spdylay_session_new(session_ptr, version, callbacks, user_data);
   if(r == 0) {
     /* IDs for use in client */
     (*session_ptr)->next_stream_id = 1;
@@ -188,11 +200,12 @@ int spdylay_session_client_new(spdylay_session **session_ptr,
 }
 
 int spdylay_session_server_new(spdylay_session **session_ptr,
+                               uint16_t version,
                                const spdylay_session_callbacks *callbacks,
                                void *user_data)
 {
   int r;
-  r = spdylay_session_new(session_ptr, callbacks, user_data);
+  r = spdylay_session_new(session_ptr, version, callbacks, user_data);
   if(r == 0) {
     (*session_ptr)->server = 1;
     /* IDs for use in client */
@@ -264,7 +277,7 @@ int spdylay_session_add_frame(spdylay_session *session,
   item->aux_data = aux_data;
   item->seq = session->next_seq++;
   /* Set priority lowest at the moment. */
-  item->pri = 3;
+  item->pri = spdylay_session_get_pri_lowest(session);
   switch(frame_type) {
   case SPDYLAY_SYN_STREAM:
     item->pri = frame->syn_stream.pri;
@@ -308,6 +321,14 @@ int spdylay_session_add_frame(spdylay_session *session,
     }
     break;
   }
+  case SPDYLAY_WINDOW_UPDATE: {
+    spdylay_stream *stream = spdylay_session_get_stream
+      (session, frame->window_update.stream_id);
+    if(stream) {
+      item->pri = stream->pri;
+    }
+    break;
+  }
   case SPDYLAY_DATA: {
     spdylay_stream *stream = spdylay_session_get_stream
       (session, frame->data.stream_id);
@@ -361,6 +382,7 @@ spdylay_stream* spdylay_session_open_stream(spdylay_session *session,
     return NULL;
   }
   spdylay_stream_init(stream, stream_id, flags, pri, initial_state,
+                      session->settings[SPDYLAY_SETTINGS_INITIAL_WINDOW_SIZE],
                       stream_user_data);
   r = spdylay_map_insert(&session->streams, stream_id, stream);
   if(r != 0) {
@@ -452,6 +474,38 @@ static int spdylay_session_is_headers_allowed(spdylay_session *session,
   }
 }
 
+/*
+ * Returns nonzero value if local endpoint can send WINDOW_UPDATE with
+ * stream ID |stream_id| at the moment.
+ */
+static int spdylay_session_is_window_update_allowed(spdylay_session *session,
+                                                    int32_t stream_id)
+{
+  spdylay_stream *stream = spdylay_session_get_stream(session, stream_id);
+  if(stream == NULL) {
+    return 0;
+  }
+  return stream->state != SPDYLAY_STREAM_CLOSING;
+}
+
+/*
+ * Returns the available window size.
+ */
+static size_t spdylay_session_avail_window(spdylay_session *session,
+                                           spdylay_stream *stream)
+{
+  if(session->flow_control == 0) {
+    return SPDYLAY_DATA_PAYLOAD_LENGTH;
+  } else {
+    if(stream->window_size >= SPDYLAY_DATA_PAYLOAD_LENGTH ||
+       stream->initial_window_size < stream->window_size*2) {
+      return stream->window_size;
+    } else {
+      return 0;
+    }
+  }
+}
+
 static int spdylay_session_is_data_allowed(spdylay_session *session,
                                            int32_t stream_id)
 {
@@ -577,6 +631,19 @@ static ssize_t spdylay_session_prep_frame(spdylay_session *session,
     }
     break;
   }
+  case SPDYLAY_WINDOW_UPDATE: {
+    if(!spdylay_session_is_window_update_allowed
+       (session, item->frame->window_update.stream_id)) {
+      return SPDYLAY_ERR_INVALID_FRAME;
+    }
+    framebuflen = spdylay_frame_pack_window_update(&session->aob.framebuf,
+                                                   &session->aob.framebufmax,
+                                                   &item->frame->window_update);
+    if(framebuflen < 0) {
+      return framebuflen;
+    }
+    break;
+  }
   case SPDYLAY_GOAWAY:
     if(session->goaway_flags & SPDYLAY_GOAWAY_SEND) {
       /* TODO The spec does not mandate that both endpoints have to
@@ -593,20 +660,28 @@ static ssize_t spdylay_session_prep_frame(spdylay_session *session,
     }
     break;
   case SPDYLAY_DATA: {
+    size_t avail_window;
+    spdylay_stream *stream;
     if(!spdylay_session_is_data_allowed(session, item->frame->data.stream_id)) {
       return SPDYLAY_ERR_INVALID_FRAME;
     }
-    framebuflen = spdylay_session_pack_data(session,
+    stream = spdylay_session_get_stream(session, item->frame->data.stream_id);
+    /* Assuming stream is not NULL */
+    assert(stream);
+    avail_window = spdylay_session_avail_window(session, stream);
+    if(avail_window == 0) {
+      spdylay_stream_defer_data(stream, item, SPDYLAY_DEFERRED_FLOW_CONTROL);
+      return SPDYLAY_ERR_DEFERRED;
+    }
+    framebuflen = spdylay_session_pack_data
+      (session,
        &session->aob.framebuf,
        &session->aob.framebufmax,
-                                            SPDYLAY_DATA_PAYLOAD_LENGTH,
+       (avail_window < SPDYLAY_DATA_PAYLOAD_LENGTH) ?
+       avail_window : SPDYLAY_DATA_PAYLOAD_LENGTH,
        &item->frame->data);
     if(framebuflen == SPDYLAY_ERR_DEFERRED) {
-      spdylay_stream *stream = spdylay_session_get_stream
-        (session, item->frame->data.stream_id);
-      /* Assuming stream is not NULL */
-      assert(stream);
-      spdylay_stream_defer_data(stream, item);
+      spdylay_stream_defer_data(stream, item, SPDYLAY_DEFERRED_NONE);
       return SPDYLAY_ERR_DEFERRED;
     } else if(framebuflen < 0) {
       return framebuflen;
@@ -826,6 +901,8 @@ static int spdylay_session_after_frame_sent(spdylay_session *session)
     }
     break;
   }
+  case SPDYLAY_WINDOW_UPDATE:
+    break;
   case SPDYLAY_DATA:
     if(frame->data.eof && (frame->data.flags & SPDYLAY_DATA_FLAG_FIN)) {
       spdylay_stream *stream =
@@ -846,21 +923,33 @@ static int spdylay_session_after_frame_sent(spdylay_session *session)
       spdylay_active_outbound_item_reset(&session->aob);
     } else {
       spdylay_outbound_item* item = spdylay_session_get_next_ob_item(session);
-      if(item == NULL || session->aob.item->pri <= item->pri) {
       /* If priority of this stream is higher or equal to other stream
          waiting at the top of the queue, we continue to send this
          data. */
-        r = spdylay_session_pack_data(session,
+      if(item == NULL || session->aob.item->pri <= item->pri) {
+        size_t avail_window;
+        spdylay_stream *stream;
+        stream = spdylay_session_get_stream(session, frame->data.stream_id);
+        /* Assuming stream is not NULL */
+        assert(stream);
+        avail_window = spdylay_session_avail_window(session, stream);
+        if(avail_window == 0) {
+          spdylay_stream_defer_data(stream, session->aob.item,
+                                    SPDYLAY_DEFERRED_FLOW_CONTROL);
+          session->aob.item = NULL;
+          spdylay_active_outbound_item_reset(&session->aob);
+          return 0;
+        }
+        r = spdylay_session_pack_data
+          (session,
            &session->aob.framebuf,
            &session->aob.framebufmax,
-                                      SPDYLAY_DATA_PAYLOAD_LENGTH,
+           (avail_window < SPDYLAY_DATA_PAYLOAD_LENGTH ?
+            avail_window : SPDYLAY_DATA_PAYLOAD_LENGTH),
            &frame->data);
         if(r == SPDYLAY_ERR_DEFERRED) {
-          spdylay_stream *stream =
-            spdylay_session_get_stream(session, frame->data.stream_id);
-          /* Assuming stream is not NULL */
-          assert(stream);
-          spdylay_stream_defer_data(stream, session->aob.item);
+          spdylay_stream_defer_data(stream, session->aob.item,
+                                    SPDYLAY_DEFERRED_NONE);
           session->aob.item = NULL;
           spdylay_active_outbound_item_reset(&session->aob);
         } else if(r < 0) {
@@ -939,6 +1028,16 @@ int spdylay_session_send(spdylay_session *session)
       }
     } else {
       session->aob.framebufoff += sentlen;
+      if(session->flow_control &&
+         session->aob.item->frame_type == SPDYLAY_DATA) {
+        spdylay_frame *frame;
+        spdylay_stream *stream;
+        frame = session->aob.item->frame;
+        stream = spdylay_session_get_stream(session, frame->data.stream_id);
+        if(stream) {
+          stream->window_size -= spdylay_get_uint32(&session->aob.framebuf[4]);
+        }
+      }
       if(session->aob.framebufoff == session->aob.framebuflen) {
         /* Frame has completely sent */
         r = spdylay_session_after_frame_sent(session);
@@ -1295,6 +1394,45 @@ int spdylay_session_on_goaway_received(spdylay_session *session,
   return 0;
 }
 
+int spdylay_session_on_window_update_received(spdylay_session *session,
+                                              spdylay_frame *frame)
+{
+  spdylay_stream *stream;
+  if(!spdylay_session_check_version(session, frame->window_update.hd.version)) {
+    return 0;
+  }
+  if(!session->flow_control) {
+    return 0;
+  }
+  stream = spdylay_session_get_stream(session, frame->window_update.stream_id);
+  if(stream) {
+    if(INT32_MAX-frame->window_update.delta_window_size < stream->window_size) {
+      int r;
+      r = spdylay_session_handle_invalid_stream
+        (session, frame->window_update.stream_id, SPDYLAY_WINDOW_UPDATE, frame,
+         SPDYLAY_FLOW_CONTROL_ERROR);
+      return r;
+    } else {
+      stream->window_size += frame->window_update.delta_window_size;
+      if(stream->deferred_data != NULL &&
+         (stream->deferred_flags & SPDYLAY_DEFERRED_FLOW_CONTROL)) {
+        int r;
+        r = spdylay_pq_push(&session->ob_pq, stream->deferred_data);
+        if(r == 0) {
+          spdylay_stream_detach_deferred_data(stream);
+        } else if(r < 0) {
+          /* FATAL */
+          assert(r < SPDYLAY_ERR_FATAL);
+          return r;
+        }
+      }
+      spdylay_session_call_on_ctrl_frame_received(session,
+                                                  SPDYLAY_WINDOW_UPDATE, frame);
+    }
+  }
+  return 0;
+}
+
 int spdylay_session_on_headers_received(spdylay_session *session,
                                         spdylay_frame *frame)
 {
@@ -1488,6 +1626,19 @@ static int spdylay_session_process_ctrl_frame(spdylay_session *session)
       r = spdylay_session_fail_session(session);
     }
     break;
+  case SPDYLAY_WINDOW_UPDATE:
+    r = spdylay_frame_unpack_window_update(&frame.window_update,
+                                           session->iframe.headbuf,
+                                           sizeof(session->iframe.headbuf),
+                                           session->iframe.buf,
+                                           session->iframe.len);
+    if(r == 0) {
+      r = spdylay_session_on_window_update_received(session, &frame);
+      spdylay_frame_window_update_free(&frame.window_update);
+    } else if(spdylay_is_non_fatal(r)) {
+      r = spdylay_session_fail_session(session);
+    }
+    break;
   }
   if(r < SPDYLAY_ERR_FATAL) {
     return r;
@@ -1568,6 +1719,40 @@ static int spdylay_session_process_data_frame(spdylay_session *session)
   }
 }
 
+/*
+ * Accumulates received bytes |delta_size| and decides whether to send
+ * WINDOW_UPDATE.
+ *
+ * This function returns 0 if it succeeds, or one of the following
+ * negative error codes:
+ *
+ * SPDYLAY_ERR_NOMEM
+ *     Out of memory.
+ */
+static int spdylay_session_update_recv_window_size(spdylay_session *session,
+                                                   int32_t stream_id,
+                                                   int32_t delta_size)
+{
+  spdylay_stream *stream;
+  stream = spdylay_session_get_stream(session, stream_id);
+  if(stream) {
+    stream->recv_window_size += delta_size;
+    /* This is just a heuristics. */
+    if(stream->recv_window_size*2 >=
+       session->settings[SPDYLAY_SETTINGS_INITIAL_WINDOW_SIZE]) {
+      int r;
+      r = spdylay_session_add_window_update(session, stream_id,
+                                            stream->recv_window_size);
+      if(r == 0) {
+        stream->recv_window_size = 0;
+      } else {
+        return r;
+      }
+    }
+  }
+  return 0;
+}
+
 int spdylay_session_recv(spdylay_session *session)
 {
   while(1) {
@@ -1616,6 +1801,8 @@ int spdylay_session_recv(spdylay_session *session)
     if(session->iframe.state == SPDYLAY_RECV_PAYLOAD) {
       size_t rempayloadlen = session->iframe.len - session->iframe.off;
       size_t bufavail, readlen;
+      int32_t data_stream_id = 0;
+      uint8_t data_flags = SPDYLAY_DATA_FLAG_NONE;
       if(spdylay_inbound_buffer_avail(&session->ibuf) == 0 &&
          rempayloadlen > 0) {
         r = spdylay_recv(session);
@@ -1632,23 +1819,39 @@ int spdylay_session_recv(spdylay_session *session)
       if(spdylay_frame_is_ctrl_frame(session->iframe.headbuf[0])) {
         memcpy(session->iframe.buf+session->iframe.off,
                session->ibuf.mark, readlen);
-      } else if(session->callbacks.on_data_chunk_recv_callback) {
-        int32_t stream_id;
-        uint8_t flags;
+      } else {
         /* For data frame, We don't buffer data. Instead, just pass
            received data to callback function. */
-        stream_id = spdylay_get_uint32(session->iframe.headbuf) &
+        data_stream_id = spdylay_get_uint32(session->iframe.headbuf) &
           SPDYLAY_STREAM_ID_MASK;
-        flags = session->iframe.headbuf[4];
+        data_flags = session->iframe.headbuf[4];
+        if(session->callbacks.on_data_chunk_recv_callback) {
           session->callbacks.on_data_chunk_recv_callback(session,
-                                                       flags,
-                                                       stream_id,
+                                                         data_flags,
+                                                         data_stream_id,
                                                          session->ibuf.mark,
                                                          readlen,
                                                          session->user_data);
         }
+      }
       session->iframe.off += readlen;
       session->ibuf.mark += readlen;
+
+      if(session->flow_control &&
+         !spdylay_frame_is_ctrl_frame(session->iframe.headbuf[0])) {
+        if(readlen > 0 &&
+           (session->iframe.len != session->iframe.off ||
+            (data_flags & SPDYLAY_DATA_FLAG_FIN) == 0)) {
+          r = spdylay_session_update_recv_window_size(session,
+                                                      data_stream_id,
+                                                      readlen);
+          if(r < 0) {
+            /* FATAL */
+            assert(r < SPDYLAY_ERR_FATAL);
+            return r;
+          }
+        }
+      }
       if(session->iframe.len == session->iframe.off) {
         if(spdylay_frame_is_ctrl_frame(session->iframe.headbuf[0])) {
           r = spdylay_session_process_ctrl_frame(session);
@@ -1738,6 +1941,26 @@ int spdylay_session_add_goaway(spdylay_session *session,
   return r;
 }
 
+int spdylay_session_add_window_update(spdylay_session *session,
+                                      int32_t stream_id,
+                                      int32_t delta_window_size)
+{
+  int r;
+  spdylay_frame *frame;
+  frame = malloc(sizeof(spdylay_frame));
+  if(frame == NULL) {
+    return SPDYLAY_ERR_NOMEM;
+  }
+  spdylay_frame_window_update_init(&frame->window_update, session->version,
+                                   stream_id, delta_window_size);
+  r = spdylay_session_add_frame(session, SPDYLAY_WINDOW_UPDATE, frame, NULL);
+  if(r != 0) {
+    spdylay_frame_window_update_free(&frame->window_update);
+    free(frame);
+  }
+  return r;
+}
+
 ssize_t spdylay_session_pack_data(spdylay_session *session,
                                   uint8_t **buf_ptr, size_t *buflen_ptr,
                                   size_t datamax,
@@ -1805,7 +2028,8 @@ int spdylay_session_resume_data(spdylay_session *session, int32_t stream_id)
   int r;
   spdylay_stream *stream;
   stream = spdylay_session_get_stream(session, stream_id);
-  if(stream == NULL || stream->deferred_data == NULL) {
+  if(stream == NULL || stream->deferred_data == NULL ||
+     (stream->deferred_flags & SPDYLAY_DEFERRED_FLOW_CONTROL)) {
     return SPDYLAY_ERR_INVALID_ARGUMENT;
   }
   r = spdylay_pq_push(&session->ob_pq, stream->deferred_data);
@@ -1814,3 +2038,14 @@ int spdylay_session_resume_data(spdylay_session *session, int32_t stream_id)
   }
   return r;
 }
+
+uint8_t spdylay_session_get_pri_lowest(spdylay_session *session)
+{
+  if(session->version == SPDYLAY_PROTO_SPDY2) {
+    return SPDYLAY_SPDY2_PRI_LOWEST;
+  } else if(session->version == SPDYLAY_PROTO_SPDY3) {
+    return SPDYLAY_SPDY3_PRI_LOWEST;
+  } else {
+    return 0;
+  }
+}

lib/spdylay_session.h

@@ -62,6 +62,8 @@ typedef struct {
   SPDYLAY_INITIAL_OUTBOUND_FRAMEBUF_LENGTH
 #define SPDYLAY_INITIAL_NV_BUFFER_LENGTH 4096
 
+#define SPDYLAY_INITIAL_WINDOW_SIZE 65536
+
 typedef struct {
   uint8_t buf[SPDYLAY_INBOUND_BUFFER_LENGTH];
   uint8_t *mark;
@@ -149,6 +151,10 @@ struct spdylay_session {
   /* This is the value in GOAWAY frame sent by remote endpoint. */
   int32_t last_good_stream_id;
 
+  /* Flag to indicate whether this session enforces flow
+     control. Nonzero for flow control enabled. */
+  uint8_t flow_control;
+
   /* Settings value store. We just use ID as index. The index = 0 is
      unused. */
   uint32_t settings[SPDYLAY_SETTINGS_MAX+1];
@@ -225,6 +231,22 @@ int spdylay_session_add_ping(spdylay_session *session, uint32_t unique_id);
 int spdylay_session_add_goaway(spdylay_session *session,
                                int32_t last_good_stream_id);
 
+/*
+ * Adds WINDOW_UPDATE frame with stream ID |stream_id| and
+ * delta-window-size |delta_window_size|. This is a convenient
+ * function built on top of spdylay_session_add_frame() to add
+ * WINDOW_UPDATE easily.
+ *
+ * This function returns 0 if it succeeds, or one of the following
+ * negative error codes:
+ *
+ * SPDYLAY_ERR_NOMEM
+ *     Out of memory.
+ */
+int spdylay_session_add_window_update(spdylay_session *session,
+                                      int32_t stream_id,
+                                      int32_t delta_window_size);
+
 /*
  * Creates new stream in |session| with stream ID |stream_id|,
  * priority |pri| and flags |flags|.  SPDYLAY_CTRL_FLAG_UNIDIRECTIONAL
@@ -363,6 +385,19 @@ int spdylay_session_on_goaway_received(spdylay_session *session,
 int spdylay_session_on_headers_received(spdylay_session *session,
                                         spdylay_frame *frame);
 
+/*
+ * Called when WINDOW_UPDATE is recieved, assuming
+ * |frame.window_update| is properly initialized.
+ *
+ * This function returns 0 if it succeeds, or one of the following
+ * negative error codes:
+ *
+ * SPDYLAY_ERR_NOMEM
+ *     Out of memory.
+ */
+int spdylay_session_on_window_update_received(spdylay_session *session,
+                                              spdylay_frame *frame);
+
 /*
  * Called when DATA is received.
  *
@@ -437,4 +472,9 @@ spdylay_outbound_item* spdylay_session_pop_next_ob_item
 spdylay_outbound_item* spdylay_session_get_next_ob_item
 (spdylay_session *session);
 
+/*
+ * Returns lowest priority value.
+ */
+uint8_t spdylay_session_get_pri_lowest(spdylay_session *session);
+
 #endif /* SPDYLAY_SESSION_H */

lib/spdylay_stream.c

@@ -29,6 +29,7 @@
 void spdylay_stream_init(spdylay_stream *stream, int32_t stream_id,
                          uint8_t flags, uint8_t pri,
                          spdylay_stream_state initial_state,
+                         int32_t initial_window_size,
                          void *stream_user_data)
 {
   stream->stream_id = stream_id;
@@ -41,6 +42,9 @@ void spdylay_stream_init(spdylay_stream *stream, int32_t stream_id,
   stream->pushed_streams_capacity = 0;
   stream->stream_user_data = stream_user_data;
   stream->deferred_data = NULL;
+  stream->deferred_flags = SPDYLAY_DEFERRED_NONE;
+  stream->initial_window_size = stream->window_size = initial_window_size;
+  stream->recv_window_size = 0;
 }
 
 void spdylay_stream_free(spdylay_stream *stream)
@@ -73,13 +77,16 @@ int spdylay_stream_add_pushed_stream(spdylay_stream *stream, int32_t stream_id)
 }
 
 void spdylay_stream_defer_data(spdylay_stream *stream,
-                               spdylay_outbound_item *data)
+                               spdylay_outbound_item *data,
+                               uint8_t flags)
 {
   assert(stream->deferred_data == NULL);
   stream->deferred_data = data;
+  stream->deferred_flags = flags;
 }
 
 void spdylay_stream_detach_deferred_data(spdylay_stream *stream)
 {
   stream->deferred_data = NULL;
+  stream->deferred_flags = SPDYLAY_DEFERRED_NONE;
 }

lib/spdylay_stream.h

@@ -69,6 +69,12 @@ typedef enum {
   SPDYLAY_SHUT_RDWR = SPDYLAY_SHUT_RD | SPDYLAY_SHUT_WR
 } spdylay_shut_flag;
 
+typedef enum {
+  SPDYLAY_DEFERRED_NONE = 0,
+  /* Indicates the DATA is deferred due to flow control. */
+  SPDYLAY_DEFERRED_FLOW_CONTROL = 0x01
+} spdylay_deferred_flag;
+
 typedef struct {
   int32_t stream_id;
   spdylay_stream_state state;
@@ -90,11 +96,27 @@ typedef struct {
   void *stream_user_data;
   /* Deferred DATA frame */
   spdylay_outbound_item *deferred_data;
+  /* The flags for defered DATA. Bitwise OR of zero or more
+     spdylay_deferred_flag values */
+  uint8_t deferred_flags;
+  /* Initial window size where window_size is compuated
+     against. Initially, window_size = initial_window_size. When N
+     bytes are sent, window_size -= N. After that, when the initial
+     window size is changed, say, new_initial_window_size, then
+     window_size becomes
+     new_initial_window_size-(initial_window_size-window_size) */
+  int32_t initial_window_size;
+  /* Current sender window size */
+  int32_t window_size;
+  /* Keep track of the number of bytes received without
+     WINDOW_UPDATE. */
+  int32_t recv_window_size;
 } spdylay_stream;
 
 void spdylay_stream_init(spdylay_stream *stream, int32_t stream_id,
                          uint8_t flags, uint8_t pri,
                          spdylay_stream_state initial_state,
+                         int32_t initial_window_size,
                          void *stream_user_data);
 
 void spdylay_stream_free(spdylay_stream *stream);
@@ -120,10 +142,12 @@ int spdylay_stream_add_pushed_stream(spdylay_stream *stream, int32_t stream_id);
 
 /*
  * Defer DATA frame |data|. We won't call this function in the
- * situation where stream->deferred_data != NULL.
+ * situation where stream->deferred_data != NULL.  If |flags| is
+ * bitwise OR of zero or more spdylay_deferred_flag values.
  */
 void spdylay_stream_defer_data(spdylay_stream *stream,
-                               spdylay_outbound_item *data);
+                               spdylay_outbound_item *data,
+                               uint8_t flags);
 
 /*
  * Detaches deferred data from this stream. This function does not

lib/spdylay_submit.c

@@ -42,7 +42,7 @@ static int spdylay_submit_syn_stream_shared
   uint8_t flags_copy;
   spdylay_data_provider *data_prd_copy = NULL;
   spdylay_syn_stream_aux_data *aux_data;
-  if(pri > 3) {
+  if(pri > spdylay_session_get_pri_lowest(session)) {
     return SPDYLAY_ERR_INVALID_ARGUMENT;
   }
   if(session->server == 0) {

tests/main.c

@@ -99,6 +99,8 @@ int main(int argc, char* argv[])
                    test_spdylay_session_reply_fail) ||
       !CU_add_test(pSuite, "session_on_headers_received",
                    test_spdylay_session_on_headers_received) ||
+      !CU_add_test(pSuite, "session_on_window_update_received",
+                   test_spdylay_session_on_window_update_received) ||
       !CU_add_test(pSuite, "session_on_ping_received",
                    test_spdylay_session_on_ping_received) ||
       !CU_add_test(pSuite, "session_on_goaway_received",
@@ -131,6 +133,8 @@ int main(int argc, char* argv[])
                    test_spdylay_session_recv_invalid_frame) ||
       !CU_add_test(pSuite, "session_defer_data",
                    test_spdylay_session_defer_data) ||
+      !CU_add_test(pSuite, "session_flow_control",
+                   test_spdylay_session_flow_control) ||
       !CU_add_test(pSuite, "frame_unpack_nv", test_spdylay_frame_unpack_nv) ||
       !CU_add_test(pSuite, "frame_count_nv_space",
                    test_spdylay_frame_count_nv_space) ||

tests/spdylay_session_test.c

@@ -198,7 +198,8 @@ void test_spdylay_session_recv()
   callbacks.recv_callback = scripted_recv_callback;
   callbacks.on_ctrl_recv_callback = on_ctrl_recv_callback;
   user_data.df = &df;
-  spdylay_session_server_new(&session, &callbacks, &user_data);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   spdylay_frame_syn_stream_init(&frame.syn_stream, SPDYLAY_PROTO_SPDY2,
                                 SPDYLAY_CTRL_FLAG_NONE,
                                 1, 0, 3, dup_nv(nv));
@@ -249,7 +250,8 @@ void test_spdylay_session_add_frame()
   memset(aux_data, 0, sizeof(spdylay_syn_stream_aux_data));
   acc.length = 0;
   user_data.acc = &acc;
-  CU_ASSERT(0 == spdylay_session_client_new(&session, &callbacks, &user_data));
+  CU_ASSERT(0 == spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, &user_data));
 
   frame = malloc(sizeof(spdylay_frame));
   spdylay_frame_syn_stream_init(&frame->syn_stream, SPDYLAY_PROTO_SPDY2,
@@ -294,7 +296,8 @@ void test_spdylay_session_recv_invalid_stream_id()
 
   user_data.df = &df;
   user_data.invalid_ctrl_recv_cb_called = 0;
-  spdylay_session_client_new(&session, &callbacks, &user_data);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   spdylay_frame_syn_stream_init(&frame.syn_stream, SPDYLAY_PROTO_SPDY2,
                                 SPDYLAY_CTRL_FLAG_NONE, 1, 0, 3, dup_nv(nv));
   framelen = spdylay_frame_pack_syn_stream(&framedata, &framedatalen,
@@ -341,7 +344,8 @@ void test_spdylay_session_on_syn_stream_received()
   user_data.ctrl_recv_cb_called = 0;
   user_data.invalid_ctrl_recv_cb_called = 0;
 
-  spdylay_session_server_new(&session, &callbacks, &user_data);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   spdylay_frame_syn_stream_init(&frame.syn_stream, SPDYLAY_PROTO_SPDY2,
                                 SPDYLAY_CTRL_FLAG_NONE,
                                 stream_id, 0, pri, dup_nv(nv));
@@ -393,7 +397,8 @@ void test_spdylay_session_on_syn_stream_received_with_push()
   user_data.ctrl_recv_cb_called = 0;
   user_data.invalid_ctrl_recv_cb_called = 0;
 
-  spdylay_session_client_new(&session, &callbacks, &user_data);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   spdylay_session_open_stream(session, assoc_stream_id, SPDYLAY_CTRL_FLAG_NONE,
                               pri, SPDYLAY_STREAM_OPENED, NULL);
   spdylay_frame_syn_stream_init(&frame.syn_stream, SPDYLAY_PROTO_SPDY2,
@@ -446,7 +451,8 @@ void test_spdylay_session_on_syn_reply_received()
   user_data.ctrl_recv_cb_called = 0;
   user_data.invalid_ctrl_recv_cb_called = 0;
 
-  spdylay_session_client_new(&session, &callbacks, &user_data);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_NONE, 0,
                               SPDYLAY_STREAM_OPENING, NULL);
   spdylay_frame_syn_reply_init(&frame.syn_reply, SPDYLAY_PROTO_SPDY2,
@@ -503,7 +509,7 @@ void test_spdylay_session_send_syn_stream()
     malloc(sizeof(spdylay_syn_stream_aux_data));
   memset(aux_data, 0, sizeof(spdylay_syn_stream_aux_data));
 
-  spdylay_session_client_new(&session, &callbacks, NULL);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, NULL);
   spdylay_frame_syn_stream_init(&frame->syn_stream, SPDYLAY_PROTO_SPDY2,
                                 SPDYLAY_CTRL_FLAG_NONE, 0, 0, 3, dup_nv(nv));
   spdylay_session_add_frame(session, SPDYLAY_SYN_STREAM, frame, aux_data);
@@ -527,7 +533,8 @@ void test_spdylay_session_send_syn_reply()
   spdylay_frame *frame = malloc(sizeof(spdylay_frame));
   spdylay_stream *stream;
 
-  CU_ASSERT(0 == spdylay_session_client_new(&session, &callbacks, NULL));
+  CU_ASSERT(0 == spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, NULL));
   spdylay_session_open_stream(session, 2, SPDYLAY_CTRL_FLAG_NONE, 3,
                               SPDYLAY_STREAM_OPENING, NULL);
   spdylay_frame_syn_reply_init(&frame->syn_reply, SPDYLAY_PROTO_SPDY2,
@@ -557,7 +564,8 @@ void test_spdylay_submit_response()
 
   data_prd.read_callback = fixed_length_data_source_read_callback;
   ud.data_source_length = 64*1024;
-  CU_ASSERT(0 == spdylay_session_client_new(&session, &callbacks, &ud));
+  CU_ASSERT(0 == spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, &ud));
   spdylay_session_open_stream(session, stream_id, SPDYLAY_CTRL_FLAG_NONE, 3,
                               SPDYLAY_STREAM_OPENING, NULL);
   CU_ASSERT(0 == spdylay_submit_response(session, stream_id, nv, &data_prd));
@@ -577,7 +585,8 @@ void test_spdylay_submit_response_with_null_data_read_callback()
 
   memset(&callbacks, 0, sizeof(callbacks));
   callbacks.send_callback = null_send_callback;
-  CU_ASSERT(0 == spdylay_session_server_new(&session, &callbacks, NULL));
+  CU_ASSERT(0 == spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, NULL));
   spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_FIN, 3,
                               SPDYLAY_STREAM_OPENING, NULL);
   CU_ASSERT(0 == spdylay_submit_response(session, 1, nv, &data_prd));
@@ -604,7 +613,8 @@ void test_spdylay_submit_request_with_data()
 
   data_prd.read_callback = fixed_length_data_source_read_callback;
   ud.data_source_length = 64*1024;
-  CU_ASSERT(0 == spdylay_session_client_new(&session, &callbacks, &ud));
+  CU_ASSERT(0 == spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, &ud));
   CU_ASSERT(0 == spdylay_submit_request(session, 3, nv, &data_prd, NULL));
   item = spdylay_session_get_next_ob_item(session);
   CU_ASSERT(0 == strcmp("version", item->frame->syn_stream.nv[0]));
@@ -627,7 +637,8 @@ void test_spdylay_submit_request_with_null_data_read_callback()
   spdylay_data_provider data_prd = {{-1}, NULL};
   spdylay_outbound_item *item;
 
-  CU_ASSERT(0 == spdylay_session_client_new(&session, &callbacks, NULL));
+  CU_ASSERT(0 == spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, NULL));
   CU_ASSERT(0 == spdylay_submit_request(session, 3, nv, &data_prd, NULL));
   item = spdylay_session_get_next_ob_item(session);
   CU_ASSERT(0 == strcmp("version", item->frame->syn_stream.nv[0]));
@@ -644,7 +655,8 @@ void test_spdylay_submit_syn_stream()
   spdylay_outbound_item *item;
 
   memset(&callbacks, 0, sizeof(spdylay_session_callbacks));
-  CU_ASSERT(0 == spdylay_session_client_new(&session, &callbacks, NULL));
+  CU_ASSERT(0 == spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, NULL));
   CU_ASSERT(0 == spdylay_submit_syn_stream(session, SPDYLAY_CTRL_FLAG_FIN, 1, 3,
                                            nv, NULL));
   item = spdylay_session_get_next_ob_item(session);
@@ -656,7 +668,8 @@ void test_spdylay_submit_syn_stream()
 
   spdylay_session_del(session);
 
-  CU_ASSERT(0 == spdylay_session_server_new(&session, &callbacks, NULL));
+  CU_ASSERT(0 == spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, NULL));
   CU_ASSERT(0 == spdylay_submit_syn_stream(session, SPDYLAY_CTRL_FLAG_FIN, 1, 3,
                                            nv, NULL));
   item = spdylay_session_get_next_ob_item(session);
@@ -681,7 +694,8 @@ void test_spdylay_submit_headers()
   callbacks.send_callback = null_send_callback;
   callbacks.on_ctrl_send_callback = on_ctrl_send_callback;
 
-  CU_ASSERT(0 == spdylay_session_client_new(&session, &callbacks, &ud));
+  CU_ASSERT(0 == spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, &ud));
   CU_ASSERT(0 == spdylay_submit_headers(session, SPDYLAY_CTRL_FLAG_FIN, 1, nv));
   item = spdylay_session_get_next_ob_item(session);
   CU_ASSERT(0 == strcmp("version", item->frame->headers.nv[0]));
@@ -720,7 +734,8 @@ void test_spdylay_session_reply_fail()
 
   data_prd.read_callback = fixed_length_data_source_read_callback;
   ud.data_source_length = 4*1024;
-  CU_ASSERT(0 == spdylay_session_client_new(&session, &callbacks, &ud));
+  CU_ASSERT(0 == spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2,
+                                            &callbacks, &ud));
   CU_ASSERT(0 == spdylay_submit_response(session, stream_id, nv, &data_prd));
   CU_ASSERT(0 == spdylay_session_send(session));
   spdylay_session_del(session);
@@ -740,7 +755,8 @@ void test_spdylay_session_on_headers_received()
   user_data.ctrl_recv_cb_called = 0;
   user_data.invalid_ctrl_recv_cb_called = 0;
 
-  spdylay_session_client_new(&session, &callbacks, &user_data);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_NONE, 0,
                               SPDYLAY_STREAM_OPENED, NULL);
   spdylay_stream_shutdown(spdylay_session_get_stream(session, 1),
@@ -805,6 +821,44 @@ void test_spdylay_session_on_headers_received()
   spdylay_session_del(session);
 }
 
+void test_spdylay_session_on_window_update_received()
+{
+  spdylay_session *session;
+  spdylay_session_callbacks callbacks;
+  my_user_data user_data;
+  spdylay_frame frame;
+  spdylay_stream *stream;
+  spdylay_outbound_item *data_item;
+  memset(&callbacks, 0, sizeof(spdylay_session_callbacks));
+  callbacks.on_ctrl_recv_callback = on_ctrl_recv_callback;
+  callbacks.on_invalid_ctrl_recv_callback = on_invalid_ctrl_recv_callback;
+  user_data.ctrl_recv_cb_called = 0;
+  user_data.invalid_ctrl_recv_cb_called = 0;
+
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY3, &callbacks,
+                             &user_data);
+  stream = spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_NONE, 0,
+                                       SPDYLAY_STREAM_OPENED, NULL);
+  spdylay_frame_window_update_init(&frame.window_update, SPDYLAY_PROTO_SPDY3,
+                                   1, 16*1024);
+
+  CU_ASSERT(0 == spdylay_session_on_window_update_received(session, &frame));
+  CU_ASSERT(1 == user_data.ctrl_recv_cb_called);
+  CU_ASSERT(64*1024+16*1024 == stream->window_size);
+
+  data_item = malloc(sizeof(spdylay_outbound_item));
+  memset(data_item, 0, sizeof(spdylay_outbound_item));
+  spdylay_stream_defer_data(stream, data_item, SPDYLAY_DEFERRED_FLOW_CONTROL);
+
+  CU_ASSERT(0 == spdylay_session_on_window_update_received(session, &frame));
+  CU_ASSERT(2 == user_data.ctrl_recv_cb_called);
+  CU_ASSERT(64*1024+16*1024*2 == stream->window_size);
+  CU_ASSERT(NULL == stream->deferred_data);
+
+  spdylay_frame_window_update_free(&frame.window_update);
+  spdylay_session_del(session);
+}
+
 void test_spdylay_session_on_ping_received()
 {
   spdylay_session *session;
@@ -821,7 +875,8 @@ void test_spdylay_session_on_ping_received()
   user_data.ctrl_recv_cb_called = 0;
   user_data.invalid_ctrl_recv_cb_called = 0;
 
-  spdylay_session_client_new(&session, &callbacks, &user_data);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   unique_id = 2;
   spdylay_frame_ping_init(&frame.ping, SPDYLAY_PROTO_SPDY2, unique_id);
 
@@ -856,7 +911,8 @@ void test_spdylay_session_on_goaway_received()
   user_data.ctrl_recv_cb_called = 0;
   user_data.invalid_ctrl_recv_cb_called = 0;
 
-  spdylay_session_client_new(&session, &callbacks, &user_data);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   spdylay_frame_goaway_init(&frame.goaway, SPDYLAY_PROTO_SPDY2, stream_id);
 
   CU_ASSERT(0 == spdylay_session_on_goaway_received(session, &frame));
@@ -877,7 +933,8 @@ void test_spdylay_session_on_data_received()
   int32_t stream_id = 2;
   spdylay_stream *stream;
 
-  spdylay_session_client_new(&session, &callbacks, &user_data);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   stream = spdylay_session_open_stream(session, stream_id,
                                        SPDYLAY_CTRL_FLAG_NONE,
                                        3, SPDYLAY_STREAM_OPENING, NULL);
@@ -920,7 +977,7 @@ void test_spdylay_session_is_my_stream_id()
   spdylay_session *session;
   spdylay_session_callbacks callbacks;
   memset(&callbacks, 0, sizeof(spdylay_session_callbacks));
-  spdylay_session_server_new(&session, &callbacks, NULL);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, NULL);
 
   CU_ASSERT(0 == spdylay_session_is_my_stream_id(session, 0));
   CU_ASSERT(0 == spdylay_session_is_my_stream_id(session, 1));
@@ -928,7 +985,7 @@ void test_spdylay_session_is_my_stream_id()
 
   spdylay_session_del(session);
 
-  spdylay_session_client_new(&session, &callbacks, NULL);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, NULL);
 
   CU_ASSERT(0 == spdylay_session_is_my_stream_id(session, 0));
   CU_ASSERT(1 == spdylay_session_is_my_stream_id(session, 1));
@@ -945,7 +1002,8 @@ void test_spdylay_session_on_rst_received()
   spdylay_stream *stream;
   spdylay_frame frame;
   memset(&callbacks, 0, sizeof(spdylay_session_callbacks));
-  spdylay_session_server_new(&session, &callbacks, &user_data);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   stream = spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_NONE,
                                        3, SPDYLAY_STREAM_OPENING, NULL);
   /* server push */
@@ -978,7 +1036,8 @@ void test_spdylay_session_send_rst_stream()
   spdylay_frame *frame;
   memset(&callbacks, 0, sizeof(spdylay_session_callbacks));
   callbacks.send_callback = null_send_callback;
-  spdylay_session_client_new(&session, &callbacks, &user_data);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   stream = spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_NONE,
                                        3, SPDYLAY_STREAM_OPENING, NULL);
   /* server push */
@@ -1010,7 +1069,7 @@ void test_spdylay_session_get_next_ob_item()
   memset(&callbacks, 0, sizeof(spdylay_session_callbacks));
   callbacks.send_callback = null_send_callback;
 
-  spdylay_session_server_new(&session, &callbacks, NULL);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, NULL);
   session->settings[SPDYLAY_SETTINGS_MAX_CONCURRENT_STREAMS] = 2;
 
   CU_ASSERT(NULL == spdylay_session_get_next_ob_item(session));
@@ -1052,7 +1111,7 @@ void test_spdylay_session_pop_next_ob_item()
   memset(&callbacks, 0, sizeof(spdylay_session_callbacks));
   callbacks.send_callback = null_send_callback;
 
-  spdylay_session_server_new(&session, &callbacks, NULL);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, NULL);
   session->settings[SPDYLAY_SETTINGS_MAX_CONCURRENT_STREAMS] = 1;
 
   CU_ASSERT(NULL == spdylay_session_pop_next_ob_item(session));
@@ -1106,7 +1165,8 @@ void test_spdylay_session_on_request_recv_callback()
   callbacks.on_request_recv_callback = on_request_recv_callback;
   user_data.stream_id = 0;
 
-  spdylay_session_server_new(&session, &callbacks, &user_data);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   spdylay_frame_syn_stream_init(&frame.syn_stream, SPDYLAY_PROTO_SPDY2,
                                 SPDYLAY_CTRL_FLAG_NONE, 1, 0, 3, dup_nv(nv));
   CU_ASSERT(0 == spdylay_session_on_syn_stream_received(session, &frame));
@@ -1174,7 +1234,8 @@ void test_spdylay_session_on_stream_close()
   callbacks.on_stream_close_callback = stream_close_callback;
   user_data.stream_close_cb_called = 0;
 
-  CU_ASSERT(spdylay_session_client_new(&session, &callbacks, &user_data) == 0);
+  CU_ASSERT(spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2,
+                                       &callbacks, &user_data) == 0);
   stream = spdylay_session_open_stream(session, stream_id,
                                        SPDYLAY_CTRL_FLAG_NONE,
                                        pri, SPDYLAY_STREAM_OPENED, &user_data);
@@ -1193,7 +1254,7 @@ void test_spdylay_session_max_concurrent_streams()
   spdylay_outbound_item *item;
 
   memset(&callbacks, 0, sizeof(spdylay_session_callbacks));
-  spdylay_session_server_new(&session, &callbacks, NULL);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, NULL);
   spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_NONE, 3,
                               SPDYLAY_STREAM_OPENED, NULL);
   spdylay_frame_syn_stream_init(&frame.syn_stream, SPDYLAY_PROTO_SPDY2,
@@ -1244,7 +1305,7 @@ void test_spdylay_session_data_backoff_by_high_pri_frame()
   ud.ctrl_send_cb_called = 0;
   ud.data_source_length = 16*1024;
 
-  spdylay_session_client_new(&session, &callbacks, &ud);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, &ud);
   spdylay_submit_request(session, 3, nv, &data_prd, NULL);
 
   ud.block_count = 2;
@@ -1289,7 +1350,7 @@ void test_spdylay_session_stop_data_with_rst_stream()
   ud.ctrl_send_cb_called = 0;
   ud.data_source_length = 16*1024;
 
-  spdylay_session_server_new(&session, &callbacks, &ud);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, &ud);
   spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_NONE, 3,
                               SPDYLAY_STREAM_OPENING, NULL);
   spdylay_submit_response(session, 1, nv, &data_prd);
@@ -1336,7 +1397,7 @@ void test_spdylay_session_stream_close_on_syn_stream()
     no_stream_user_data_stream_close_callback;
   ud.stream_close_cb_called = 0;
 
-  spdylay_session_client_new(&session, &callbacks, &ud);
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, &ud);
   spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_NONE, 3,
                               SPDYLAY_STREAM_OPENING, NULL);
   spdylay_frame_syn_stream_init(&frame.syn_stream, SPDYLAY_PROTO_SPDY2,
@@ -1371,7 +1432,8 @@ void test_spdylay_session_recv_invalid_frame()
 
   user_data.df = &df;
   user_data.ctrl_send_cb_called = 0;
-  spdylay_session_server_new(&session, &callbacks, &user_data);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks,
+                             &user_data);
   spdylay_frame_syn_stream_init(&frame.syn_stream, SPDYLAY_PROTO_SPDY2,
                                 SPDYLAY_CTRL_FLAG_NONE, 1, 0, 3, dup_nv(nv));
   framelen = spdylay_frame_pack_syn_stream(&framedata, &framedatalen,
@@ -1424,7 +1486,7 @@ void test_spdylay_session_defer_data()
   ud.ctrl_send_cb_called = 0;
   ud.data_source_length = 16*1024;
 
-  spdylay_session_server_new(&session, &callbacks, &ud);
+  spdylay_session_server_new(&session, SPDYLAY_PROTO_SPDY2, &callbacks, &ud);
   spdylay_session_open_stream(session, 1, SPDYLAY_CTRL_FLAG_NONE, 3,
                               SPDYLAY_STREAM_OPENING, NULL);
   spdylay_submit_response(session, 1, nv, &data_prd);
@@ -1473,3 +1535,47 @@ void test_spdylay_session_defer_data()
 
   spdylay_session_del(session);
 }
+
+void test_spdylay_session_flow_control()
+{
+  spdylay_session *session;
+  spdylay_session_callbacks callbacks;
+  const char *nv[] = { NULL };
+  my_user_data ud;
+  spdylay_data_provider data_prd;
+  spdylay_frame frame;
+
+  memset(&callbacks, 0, sizeof(spdylay_session_callbacks));
+  callbacks.send_callback = null_send_callback;
+  callbacks.on_ctrl_send_callback = on_ctrl_send_callback;
+  data_prd.read_callback = fixed_length_data_source_read_callback;
+
+  ud.ctrl_send_cb_called = 0;
+  ud.data_source_length = 128*1024;
+
+  spdylay_session_client_new(&session, SPDYLAY_PROTO_SPDY3, &callbacks, &ud);
+  spdylay_submit_request(session, 3, nv, &data_prd, NULL);
+
+  /* Sends 64KiB data */
+  CU_ASSERT(0 == spdylay_session_send(session));
+  CU_ASSERT(64*1024 == ud.data_source_length);
+
+  spdylay_frame_window_update_init(&frame.window_update, SPDYLAY_PROTO_SPDY3,
+                                   1, 32*1024);
+  spdylay_session_on_window_update_received(session, &frame);
+
+  /* Sends another 32KiB data */
+  CU_ASSERT(0 == spdylay_session_send(session));
+  CU_ASSERT(32*1024 == ud.data_source_length);
+
+  spdylay_session_on_window_update_received(session, &frame);
+
+  /* Sends another 32KiB data */
+  CU_ASSERT(0 == spdylay_session_send(session));
+  CU_ASSERT(0 == ud.data_source_length);
+  CU_ASSERT(spdylay_session_get_stream(session, 1)->shut_flags &
+            SPDYLAY_SHUT_WR);
+
+  spdylay_frame_window_update_free(&frame.window_update);
+  spdylay_session_del(session);
+}

tests/spdylay_session_test.h

@@ -31,6 +31,7 @@ void test_spdylay_session_add_frame();
 void test_spdylay_session_on_syn_stream_received();
 void test_spdylay_session_on_syn_stream_received_with_push();
 void test_spdylay_session_on_syn_reply_received();
+void test_spdylay_session_on_window_update_received();
 void test_spdylay_session_send_syn_stream();
 void test_spdylay_session_send_syn_reply();
 void test_spdylay_submit_response();
@@ -57,5 +58,6 @@ void test_spdylay_session_stop_data_with_rst_stream();
 void test_spdylay_session_stream_close_on_syn_stream();
 void test_spdylay_session_recv_invalid_frame();
 void test_spdylay_session_defer_data();
+void test_spdylay_session_flow_control();
 
 #endif // SPDYLAY_SESSION_TEST_H

tests/spdylay_stream_test.c

@@ -32,7 +32,7 @@ void test_spdylay_stream_add_pushed_stream()
 {
   spdylay_stream stream;
   int i, n;
-  spdylay_stream_init(&stream, 1, SPDYLAY_CTRL_FLAG_NONE, 3,
+  spdylay_stream_init(&stream, 1, SPDYLAY_CTRL_FLAG_NONE, 3, 65536,
                       SPDYLAY_STREAM_OPENING, NULL);
   n = 26;
   for(i = 2; i < n; i += 2) {