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