lib/stream.c

   1 /*
   2  * Copyright (c) 2008, 2009, 2010 Nicira Networks.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at:
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include <config.h>
  18 #include "stream-provider.h"
  19 #include <assert.h>
  20 #include <errno.h>
  21 #include <inttypes.h>
  22 #include <netinet/in.h>
  23 #include <poll.h>
  24 #include <stdlib.h>
  25 #include <string.h>
  26 #include "coverage.h"
  27 #include "dynamic-string.h"
  28 #include "fatal-signal.h"
  29 #include "flow.h"
  30 #include "ofp-print.h"
  31 #include "ofpbuf.h"
  32 #include "openflow/nicira-ext.h"
  33 #include "openflow/openflow.h"
  34 #include "packets.h"
  35 #include "poll-loop.h"
  36 #include "random.h"
  37 #include "util.h"
  38 #include "vlog.h"
  39
  40 VLOG_DEFINE_THIS_MODULE(stream)
  41
  42 /* State of an active stream.*/
  43 enum stream_state {
  44     SCS_CONNECTING,             /* Underlying stream is not connected. */
  45     SCS_CONNECTED,              /* Connection established. */
  46     SCS_DISCONNECTED            /* Connection failed or connection closed. */
  47 };
  48
  49 static struct stream_class *stream_classes[] = {
  50     &tcp_stream_class,
  51     &unix_stream_class,
  52 #ifdef HAVE_OPENSSL
  53     &ssl_stream_class,
  54 #endif
  55 };
  56
  57 static struct pstream_class *pstream_classes[] = {
  58     &ptcp_pstream_class,
  59     &punix_pstream_class,
  60 #ifdef HAVE_OPENSSL
  61     &pssl_pstream_class,
  62 #endif
  63 };
  64
  65 /* Check the validity of the stream class structures. */
  66 static void
  67 check_stream_classes(void)
  68 {
  69 #ifndef NDEBUG
  70     size_t i;
  71
  72     for (i = 0; i < ARRAY_SIZE(stream_classes); i++) {
  73         struct stream_class *class = stream_classes[i];
  74         assert(class->name != NULL);
  75         assert(class->open != NULL);
  76         if (class->close || class->recv || class->send || class->run
  77             || class->run_wait || class->wait) {
  78             assert(class->close != NULL);
  79             assert(class->recv != NULL);
  80             assert(class->send != NULL);
  81             assert(class->wait != NULL);
  82         } else {
  83             /* This class delegates to another one. */
  84         }
  85     }
  86
  87     for (i = 0; i < ARRAY_SIZE(pstream_classes); i++) {
  88         struct pstream_class *class = pstream_classes[i];
  89         assert(class->name != NULL);
  90         assert(class->listen != NULL);
  91         if (class->close || class->accept || class->wait) {
  92             assert(class->close != NULL);
  93             assert(class->accept != NULL);
  94             assert(class->wait != NULL);
  95         } else {
  96             /* This class delegates to another one. */
  97         }
  98     }
  99 #endif
 100 }
 101
 102 /* Prints information on active (if 'active') and passive (if 'passive')
 103  * connection methods supported by the stream. */
 104 void
 105 stream_usage(const char *name, bool active, bool passive,
 106              bool bootstrap OVS_UNUSED)
 107 {
 108     /* Really this should be implemented via callbacks into the stream
 109      * providers, but that seems too heavy-weight to bother with at the
 110      * moment. */
 111
 112     printf("\n");
 113     if (active) {
 114         printf("Active %s connection methods:\n", name);
 115         printf("  tcp:IP:PORT             "
 116                "PORT at remote IP\n");
 117 #ifdef HAVE_OPENSSL
 118         printf("  ssl:IP:PORT             "
 119                "SSL PORT at remote IP\n");
 120 #endif
 121         printf("  unix:FILE               "
 122                "Unix domain socket named FILE\n");
 123     }
 124
 125     if (passive) {
 126         printf("Passive %s connection methods:\n", name);
 127         printf("  ptcp:PORT[:IP]          "
 128                "listen to TCP PORT on IP\n");
 129 #ifdef HAVE_OPENSSL
 130         printf("  pssl:PORT[:IP]          "
 131                "listen for SSL on PORT on IP\n");
 132 #endif
 133         printf("  punix:FILE              "
 134                "listen on Unix domain socket FILE\n");
 135     }
 136
 137 #ifdef HAVE_OPENSSL
 138     printf("PKI configuration (required to use SSL):\n"
 139            "  -p, --private-key=FILE  file with private key\n"
 140            "  -c, --certificate=FILE  file with certificate for private key\n"
 141            "  -C, --ca-cert=FILE      file with peer CA certificate\n");
 142     if (bootstrap) {
 143         printf("  --bootstrap-ca-cert=FILE  file with peer CA certificate "
 144                "to read or create\n");
 145     }
 146 #endif
 147 }
 148
 149 /* Given 'name', a stream name in the form "TYPE:ARGS", stores the class
 150  * named "TYPE" into '*classp' and returns 0.  Returns EAFNOSUPPORT and stores
 151  * a null pointer into '*classp' if 'name' is in the wrong form or if no such
 152  * class exists. */
 153 static int
 154 stream_lookup_class(const char *name, struct stream_class **classp)
 155 {
 156     size_t prefix_len;
 157     size_t i;
 158
 159     check_stream_classes();
 160
 161     *classp = NULL;
 162     prefix_len = strcspn(name, ":");
 163     if (name[prefix_len] == '\0') {
 164         return EAFNOSUPPORT;
 165     }
 166     for (i = 0; i < ARRAY_SIZE(stream_classes); i++) {
 167         struct stream_class *class = stream_classes[i];
 168         if (strlen(class->name) == prefix_len
 169             && !memcmp(class->name, name, prefix_len)) {
 170             *classp = class;
 171             return 0;
 172         }
 173     }
 174     return EAFNOSUPPORT;
 175 }
 176
 177 /* Returns 0 if 'name' is a stream name in the form "TYPE:ARGS" and TYPE is
 178  * a supported stream type, otherwise EAFNOSUPPORT.  */
 179 int
 180 stream_verify_name(const char *name)
 181 {
 182     struct stream_class *class;
 183     return stream_lookup_class(name, &class);
 184 }
 185
 186 /* Attempts to connect a stream to a remote peer.  'name' is a connection name
 187  * in the form "TYPE:ARGS", where TYPE is an active stream class's name and
 188  * ARGS are stream class-specific.
 189  *
 190  * Returns 0 if successful, otherwise a positive errno value.  If successful,
 191  * stores a pointer to the new connection in '*streamp', otherwise a null
 192  * pointer.  */
 193 int
 194 stream_open(const char *name, struct stream **streamp)
 195 {
 196     struct stream_class *class;
 197     struct stream *stream;
 198     char *suffix_copy;
 199     int error;
 200
 201     COVERAGE_INC(stream_open);
 202
 203     /* Look up the class. */
 204     error = stream_lookup_class(name, &class);
 205     if (!class) {
 206         goto error;
 207     }
 208
 209     /* Call class's "open" function. */
 210     suffix_copy = xstrdup(strchr(name, ':') + 1);
 211     error = class->open(name, suffix_copy, &stream);
 212     free(suffix_copy);
 213     if (error) {
 214         goto error;
 215     }
 216
 217     /* Success. */
 218     *streamp = stream;
 219     return 0;
 220
 221 error:
 222     *streamp = NULL;
 223     return error;
 224 }
 225
 226 /* Blocks until a previously started stream connection attempt succeeds or
 227  * fails.  'error' should be the value returned by stream_open() and 'streamp'
 228  * should point to the stream pointer set by stream_open().  Returns 0 if
 229  * successful, otherwise a positive errno value other than EAGAIN or
 230  * EINPROGRESS.  If successful, leaves '*streamp' untouched; on error, closes
 231  * '*streamp' and sets '*streamp' to null.
 232  *
 233  * Typical usage:
 234  *   error = stream_open_block(stream_open("tcp:1.2.3.4:5", &stream), &stream);
 235  */
 236 int
 237 stream_open_block(int error, struct stream **streamp)
 238 {
 239     struct stream *stream = *streamp;
 240
 241     fatal_signal_run();
 242
 243     if (!error) {
 244         while ((error = stream_connect(stream)) == EAGAIN) {
 245             stream_run(stream);
 246             stream_run_wait(stream);
 247             stream_connect_wait(stream);
 248             poll_block();
 249         }
 250         assert(error != EINPROGRESS);
 251     }
 252
 253     if (error) {
 254         stream_close(stream);
 255         *streamp = NULL;
 256     } else {
 257         *streamp = stream;
 258     }
 259     return error;
 260 }
 261
 262 /* Closes 'stream'. */
 263 void
 264 stream_close(struct stream *stream)
 265 {
 266     if (stream != NULL) {
 267         char *name = stream->name;
 268         (stream->class->close)(stream);
 269         free(name);
 270     }
 271 }
 272
 273 /* Returns the name of 'stream', that is, the string passed to
 274  * stream_open(). */
 275 const char *
 276 stream_get_name(const struct stream *stream)
 277 {
 278     return stream ? stream->name : "(null)";
 279 }
 280
 281 /* Returns the IP address of the peer, or 0 if the peer is not connected over
 282  * an IP-based protocol or if its IP address is not yet known. */
 283 uint32_t
 284 stream_get_remote_ip(const struct stream *stream)
 285 {
 286     return stream->remote_ip;
 287 }
 288
 289 /* Returns the transport port of the peer, or 0 if the connection does not
 290  * contain a port or if the port is not yet known. */
 291 uint16_t
 292 stream_get_remote_port(const struct stream *stream)
 293 {
 294     return stream->remote_port;
 295 }
 296
 297 /* Returns the IP address used to connect to the peer, or 0 if the connection
 298  * is not an IP-based protocol or if its IP address is not yet known. */
 299 uint32_t
 300 stream_get_local_ip(const struct stream *stream)
 301 {
 302     return stream->local_ip;
 303 }
 304
 305 /* Returns the transport port used to connect to the peer, or 0 if the
 306  * connection does not contain a port or if the port is not yet known. */
 307 uint16_t
 308 stream_get_local_port(const struct stream *stream)
 309 {
 310     return stream->local_port;
 311 }
 312
 313 static void
 314 scs_connecting(struct stream *stream)
 315 {
 316     int retval = (stream->class->connect)(stream);
 317     assert(retval != EINPROGRESS);
 318     if (!retval) {
 319         stream->state = SCS_CONNECTED;
 320     } else if (retval != EAGAIN) {
 321         stream->state = SCS_DISCONNECTED;
 322         stream->error = retval;
 323     }
 324 }
 325
 326 /* Tries to complete the connection on 'stream'.  If 'stream''s connection is
 327  * complete, returns 0 if the connection was successful or a positive errno
 328  * value if it failed.  If the connection is still in progress, returns
 329  * EAGAIN. */
 330 int
 331 stream_connect(struct stream *stream)
 332 {
 333     enum stream_state last_state;
 334
 335     do {
 336         last_state = stream->state;
 337         switch (stream->state) {
 338         case SCS_CONNECTING:
 339             scs_connecting(stream);
 340             break;
 341
 342         case SCS_CONNECTED:
 343             return 0;
 344
 345         case SCS_DISCONNECTED:
 346             return stream->error;
 347
 348         default:
 349             NOT_REACHED();
 350         }
 351     } while (stream->state != last_state);
 352
 353     return EAGAIN;
 354 }
 355
 356 /* Tries to receive up to 'n' bytes from 'stream' into 'buffer', and returns:
 357  *
 358  *     - If successful, the number of bytes received (between 1 and 'n').
 359  *
 360  *     - On error, a negative errno value.
 361  *
 362  *     - 0, if the connection has been closed in the normal fashion, or if 'n'
 363  *       is zero.
 364  *
 365  * The recv function will not block waiting for a packet to arrive.  If no
 366  * data have been received, it returns -EAGAIN immediately. */
 367 int
 368 stream_recv(struct stream *stream, void *buffer, size_t n)
 369 {
 370     int retval = stream_connect(stream);
 371     return (retval ? -retval
 372             : n == 0 ? 0
 373             : (stream->class->recv)(stream, buffer, n));
 374 }
 375
 376 /* Tries to send up to 'n' bytes of 'buffer' on 'stream', and returns:
 377  *
 378  *     - If successful, the number of bytes sent (between 1 and 'n').  0 is
 379  *       only a valid return value if 'n' is 0.
 380  *
 381  *     - On error, a negative errno value.
 382  *
 383  * The send function will not block.  If no bytes can be immediately accepted
 384  * for transmission, it returns -EAGAIN immediately. */
 385 int
 386 stream_send(struct stream *stream, const void *buffer, size_t n)
 387 {
 388     int retval = stream_connect(stream);
 389     return (retval ? -retval
 390             : n == 0 ? 0
 391             : (stream->class->send)(stream, buffer, n));
 392 }
 393
 394 /* Allows 'stream' to perform maintenance activities, such as flushing
 395  * output buffers. */
 396 void
 397 stream_run(struct stream *stream)
 398 {
 399     if (stream->class->run) {
 400         (stream->class->run)(stream);
 401     }
 402 }
 403
 404 /* Arranges for the poll loop to wake up when 'stream' needs to perform
 405  * maintenance activities. */
 406 void
 407 stream_run_wait(struct stream *stream)
 408 {
 409     if (stream->class->run_wait) {
 410         (stream->class->run_wait)(stream);
 411     }
 412 }
 413
 414 /* Arranges for the poll loop to wake up when 'stream' is ready to take an
 415  * action of the given 'type'. */
 416 void
 417 stream_wait(struct stream *stream, enum stream_wait_type wait)
 418 {
 419     assert(wait == STREAM_CONNECT || wait == STREAM_RECV
 420            || wait == STREAM_SEND);
 421
 422     switch (stream->state) {
 423     case SCS_CONNECTING:
 424         wait = STREAM_CONNECT;
 425         break;
 426
 427     case SCS_DISCONNECTED:
 428         poll_immediate_wake();
 429         return;
 430     }
 431     (stream->class->wait)(stream, wait);
 432 }
 433
 434 void
 435 stream_connect_wait(struct stream *stream)
 436 {
 437     stream_wait(stream, STREAM_CONNECT);
 438 }
 439
 440 void
 441 stream_recv_wait(struct stream *stream)
 442 {
 443     stream_wait(stream, STREAM_RECV);
 444 }
 445
 446 void
 447 stream_send_wait(struct stream *stream)
 448 {
 449     stream_wait(stream, STREAM_SEND);
 450 }
 451
 452 /* Given 'name', a pstream name in the form "TYPE:ARGS", stores the class
 453  * named "TYPE" into '*classp' and returns 0.  Returns EAFNOSUPPORT and stores
 454  * a null pointer into '*classp' if 'name' is in the wrong form or if no such
 455  * class exists. */
 456 static int
 457 pstream_lookup_class(const char *name, struct pstream_class **classp)
 458 {
 459     size_t prefix_len;
 460     size_t i;
 461
 462     check_stream_classes();
 463
 464     *classp = NULL;
 465     prefix_len = strcspn(name, ":");
 466     if (name[prefix_len] == '\0') {
 467         return EAFNOSUPPORT;
 468     }
 469     for (i = 0; i < ARRAY_SIZE(pstream_classes); i++) {
 470         struct pstream_class *class = pstream_classes[i];
 471         if (strlen(class->name) == prefix_len
 472             && !memcmp(class->name, name, prefix_len)) {
 473             *classp = class;
 474             return 0;
 475         }
 476     }
 477     return EAFNOSUPPORT;
 478 }
 479
 480 /* Returns 0 if 'name' is a pstream name in the form "TYPE:ARGS" and TYPE is
 481  * a supported pstream type, otherwise EAFNOSUPPORT.  */
 482 int
 483 pstream_verify_name(const char *name)
 484 {
 485     struct pstream_class *class;
 486     return pstream_lookup_class(name, &class);
 487 }
 488
 489 /* Attempts to start listening for remote stream connections.  'name' is a
 490  * connection name in the form "TYPE:ARGS", where TYPE is an passive stream
 491  * class's name and ARGS are stream class-specific.
 492  *
 493  * Returns 0 if successful, otherwise a positive errno value.  If successful,
 494  * stores a pointer to the new connection in '*pstreamp', otherwise a null
 495  * pointer.  */
 496 int
 497 pstream_open(const char *name, struct pstream **pstreamp)
 498 {
 499     struct pstream_class *class;
 500     struct pstream *pstream;
 501     char *suffix_copy;
 502     int error;
 503
 504     COVERAGE_INC(pstream_open);
 505
 506     /* Look up the class. */
 507     error = pstream_lookup_class(name, &class);
 508     if (!class) {
 509         goto error;
 510     }
 511
 512     /* Call class's "open" function. */
 513     suffix_copy = xstrdup(strchr(name, ':') + 1);
 514     error = class->listen(name, suffix_copy, &pstream);
 515     free(suffix_copy);
 516     if (error) {
 517         goto error;
 518     }
 519
 520     /* Success. */
 521     *pstreamp = pstream;
 522     return 0;
 523
 524 error:
 525     *pstreamp = NULL;
 526     return error;
 527 }
 528
 529 /* Returns the name that was used to open 'pstream'.  The caller must not
 530  * modify or free the name. */
 531 const char *
 532 pstream_get_name(const struct pstream *pstream)
 533 {
 534     return pstream->name;
 535 }
 536
 537 /* Closes 'pstream'. */
 538 void
 539 pstream_close(struct pstream *pstream)
 540 {
 541     if (pstream != NULL) {
 542         char *name = pstream->name;
 543         (pstream->class->close)(pstream);
 544         free(name);
 545     }
 546 }
 547
 548 /* Tries to accept a new connection on 'pstream'.  If successful, stores the
 549  * new connection in '*new_stream' and returns 0.  Otherwise, returns a
 550  * positive errno value.
 551  *
 552  * pstream_accept() will not block waiting for a connection.  If no connection
 553  * is ready to be accepted, it returns EAGAIN immediately. */
 554 int
 555 pstream_accept(struct pstream *pstream, struct stream **new_stream)
 556 {
 557     int retval = (pstream->class->accept)(pstream, new_stream);
 558     if (retval) {
 559         *new_stream = NULL;
 560     } else {
 561         assert((*new_stream)->state != SCS_CONNECTING
 562                || (*new_stream)->class->connect);
 563     }
 564     return retval;
 565 }
 566
 567 /* Tries to accept a new connection on 'pstream'.  If successful, stores the
 568  * new connection in '*new_stream' and returns 0.  Otherwise, returns a
 569  * positive errno value.
 570  *
 571  * pstream_accept_block() blocks until a connection is ready or until an error
 572  * occurs.  It will not return EAGAIN. */
 573 int
 574 pstream_accept_block(struct pstream *pstream, struct stream **new_stream)
 575 {
 576     int error;
 577
 578     fatal_signal_run();
 579     while ((error = pstream_accept(pstream, new_stream)) == EAGAIN) {
 580         pstream_wait(pstream);
 581         poll_block();
 582     }
 583     if (error) {
 584         *new_stream = NULL;
 585     }
 586     return error;
 587 }
 588
 589 void
 590 pstream_wait(struct pstream *pstream)
 591 {
 592     (pstream->class->wait)(pstream);
 593 }
 594 \f
 595 /* Initializes 'stream' as a new stream named 'name', implemented via 'class'.
 596  * The initial connection status, supplied as 'connect_status', is interpreted
 597  * as follows:
 598  *
 599  *      - 0: 'stream' is connected.  Its 'send' and 'recv' functions may be
 600  *        called in the normal fashion.
 601  *
 602  *      - EAGAIN: 'stream' is trying to complete a connection.  Its 'connect'
 603  *        function should be called to complete the connection.
 604  *
 605  *      - Other positive errno values indicate that the connection failed with
 606  *        the specified error.
 607  *
 608  * After calling this function, stream_close() must be used to destroy
 609  * 'stream', otherwise resources will be leaked.
 610  *
 611  * The caller retains ownership of 'name'. */
 612 void
 613 stream_init(struct stream *stream, struct stream_class *class,
 614             int connect_status, const char *name)
 615 {
 616     stream->class = class;
 617     stream->state = (connect_status == EAGAIN ? SCS_CONNECTING
 618                     : !connect_status ? SCS_CONNECTED
 619                     : SCS_DISCONNECTED);
 620     stream->error = connect_status;
 621     stream->name = xstrdup(name);
 622     assert(stream->state != SCS_CONNECTING || class->connect);
 623 }
 624
 625 void
 626 stream_set_remote_ip(struct stream *stream, uint32_t ip)
 627 {
 628     stream->remote_ip = ip;
 629 }
 630
 631 void
 632 stream_set_remote_port(struct stream *stream, uint16_t port)
 633 {
 634     stream->remote_port = port;
 635 }
 636
 637 void
 638 stream_set_local_ip(struct stream *stream, uint32_t ip)
 639 {
 640     stream->local_ip = ip;
 641 }
 642
 643 void
 644 stream_set_local_port(struct stream *stream, uint16_t port)
 645 {
 646     stream->local_port = port;
 647 }
 648
 649 void
 650 pstream_init(struct pstream *pstream, struct pstream_class *class,
 651             const char *name)
 652 {
 653     pstream->class = class;
 654     pstream->name = xstrdup(name);
 655 }
 656 \f
 657 static int
 658 count_fields(const char *s_)
 659 {
 660     char *s, *field, *save_ptr;
 661     int n = 0;
 662
 663     save_ptr = NULL;
 664     s = xstrdup(s_);
 665     for (field = strtok_r(s, ":", &save_ptr); field != NULL;
 666          field = strtok_r(NULL, ":", &save_ptr)) {
 667         n++;
 668     }
 669     free(s);
 670
 671     return n;
 672 }
 673
 674 /* Like stream_open(), but for tcp streams the port defaults to
 675  * 'default_tcp_port' if no port number is given and for SSL streams the port
 676  * defaults to 'default_ssl_port' if no port number is given. */
 677 int
 678 stream_open_with_default_ports(const char *name_,
 679                                uint16_t default_tcp_port,
 680                                uint16_t default_ssl_port,
 681                                struct stream **streamp)
 682 {
 683     char *name;
 684     int error;
 685
 686     if (!strncmp(name_, "tcp:", 4) && count_fields(name_) < 3) {
 687         name = xasprintf("%s:%d", name_, default_tcp_port);
 688     } else if (!strncmp(name_, "ssl:", 4) && count_fields(name_) < 3) {
 689         name = xasprintf("%s:%d", name_, default_ssl_port);
 690     } else {
 691         name = xstrdup(name_);
 692     }
 693     error = stream_open(name, streamp);
 694     free(name);
 695
 696     return error;
 697 }
 698
 699 /* Like pstream_open(), but for ptcp streams the port defaults to
 700  * 'default_ptcp_port' if no port number is given and for passive SSL streams
 701  * the port defaults to 'default_pssl_port' if no port number is given. */
 702 int
 703 pstream_open_with_default_ports(const char *name_,
 704                                 uint16_t default_ptcp_port,
 705                                 uint16_t default_pssl_port,
 706                                 struct pstream **pstreamp)
 707 {
 708     char *name;
 709     int error;
 710
 711     if (!strncmp(name_, "ptcp:", 5) && count_fields(name_) < 2) {
 712         name = xasprintf("%s%d", name_, default_ptcp_port);
 713     } else if (!strncmp(name_, "pssl:", 5) && count_fields(name_) < 2) {
 714         name = xasprintf("%s%d", name_, default_pssl_port);
 715     } else {
 716         name = xstrdup(name_);
 717     }
 718     error = pstream_open(name, pstreamp);
 719     free(name);
 720
 721     return error;
 722 }
 723 \f
 724 /* Attempts to guess the content type of a stream whose first few bytes were
 725  * the 'size' bytes of 'data'. */
 726 static enum stream_content_type
 727 stream_guess_content(const uint8_t *data, size_t size)
 728 {
 729     if (size >= 2) {
 730 #define PAIR(A, B) (((A) << 8) | (B))
 731         switch (PAIR(data[0], data[1])) {
 732         case PAIR(0x16, 0x03):  /* Handshake, version 3. */
 733             return STREAM_SSL;
 734         case PAIR('{', '"'):
 735             return STREAM_JSONRPC;
 736         case PAIR(OFP_VERSION, OFPT_HELLO):
 737             return STREAM_OPENFLOW;
 738         }
 739     }
 740
 741     return STREAM_UNKNOWN;
 742 }
 743
 744 /* Returns a string represenation of 'type'. */
 745 static const char *
 746 stream_content_type_to_string(enum stream_content_type type)
 747 {
 748     switch (type) {
 749     case STREAM_UNKNOWN:
 750     default:
 751         return "unknown";
 752
 753     case STREAM_JSONRPC:
 754         return "JSON-RPC";
 755
 756     case STREAM_OPENFLOW:
 757         return "OpenFlow";
 758
 759     case STREAM_SSL:
 760         return "SSL";
 761     }
 762 }
 763
 764 /* Attempts to guess the content type of a stream whose first few bytes were
 765  * the 'size' bytes of 'data'.  If this is done successfully, and the guessed
 766  * content type is other than 'expected_type', then log a message in vlog
 767  * module 'module', naming 'stream_name' as the source, explaining what
 768  * content was expected and what was actually received. */
 769 void
 770 stream_report_content(const void *data, size_t size,
 771                       enum stream_content_type expected_type,
 772                       struct vlog_module *module, const char *stream_name)
 773 {
 774     static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(5, 5);
 775     enum stream_content_type actual_type;
 776
 777     actual_type = stream_guess_content(data, size);
 778     if (actual_type != expected_type && actual_type != STREAM_UNKNOWN) {
 779         vlog_rate_limit(module, VLL_WARN, &rl,
 780                         "%s: received %s data on %s channel",
 781                         stream_name,
 782                         stream_content_type_to_string(expected_type),
 783                         stream_content_type_to_string(actual_type));
 784     }
 785 }