lib/stream.c

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