4 files changed, 58 insertions, 29 deletions

diff --git a/librabbitmq/amqp-openssl.c b/librabbitmq/amqp-openssl.c
index 16d81e0..4678a4a 100644
--- a/librabbitmq/amqp-openssl.c
+++ b/librabbitmq/amqp-openssl.c
@@ -55,7 +55,7 @@ static pthread_mutex_t *amqp_openssl_lockarray = NULL;
 #endif /* ENABLE_THREAD_SAFETY */
 
 struct amqp_ssl_socket_t {
-	amqp_socket_t base;
+	const struct amqp_socket_class_t *klass;
 	BIO *bio;
 	SSL_CTX *ctx;
 	char *buffer;
@@ -269,6 +269,16 @@ amqp_ssl_socket_get_sockfd(void *base)
 	return BIO_get_fd(self->bio, NULL);
 }
 
+static const struct amqp_socket_class_t amqp_ssl_socket_class = {
+	amqp_ssl_socket_writev, /* writev */
+	amqp_ssl_socket_send, /* send */
+	amqp_ssl_socket_recv, /* recv */
+	amqp_ssl_socket_open, /* open */
+	amqp_ssl_socket_close, /* close */
+	amqp_ssl_socket_error, /* error */
+	amqp_ssl_socket_get_sockfd /* get_sockfd */
+};
+
 amqp_socket_t *
 amqp_ssl_socket_new(void)
 {
@@ -285,13 +295,7 @@ amqp_ssl_socket_new(void)
 	if (!self->ctx) {
 		goto error;
 	}
-	self->base.writev = amqp_ssl_socket_writev;
-	self->base.send = amqp_ssl_socket_send;
-	self->base.recv = amqp_ssl_socket_recv;
-	self->base.open = amqp_ssl_socket_open;
-	self->base.close = amqp_ssl_socket_close;
-	self->base.error = amqp_ssl_socket_error;
-	self->base.get_sockfd = amqp_ssl_socket_get_sockfd;
+	self->klass = &amqp_ssl_socket_class;
 	self->verify = 1;
 	return (amqp_socket_t *)self;
 error:
@@ -303,8 +307,13 @@ int
 amqp_ssl_socket_set_cacert(amqp_socket_t *base,
 			   const char *cacert)
 {
-	struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
-	int status = SSL_CTX_load_verify_locations(self->ctx, cacert, NULL);
+	int status;
+	struct amqp_ssl_socket_t *self;
+	if (base->klass != &amqp_ssl_socket_class) {
+		amqp_abort("<%p> is not of type amqp_ssl_socket_t", base);
+	}
+	self = (struct amqp_ssl_socket_t *)base;
+	status = SSL_CTX_load_verify_locations(self->ctx, cacert, NULL);
 	if (1 != status) {
 		return -1;
 	}
@@ -316,7 +325,11 @@ amqp_ssl_socket_set_key(amqp_socket_t *base,
 			const char *key,
 			const char *cert)
 {
-	struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
+	struct amqp_ssl_socket_t *self;
+	if (base->klass != &amqp_ssl_socket_class) {
+		amqp_abort("<%p> is not of type amqp_ssl_socket_t", base);
+	}
+	self = (struct amqp_ssl_socket_t *)base;
 	if (key && cert) {
 		int status = SSL_CTX_use_PrivateKey_file(self->ctx, key,
 							 SSL_FILETYPE_PEM);
@@ -336,7 +349,11 @@ void
 amqp_ssl_socket_set_verify(amqp_socket_t *base,
 			   amqp_boolean_t verify)
 {
-	struct amqp_ssl_socket_t *self = (struct amqp_ssl_socket_t *)base;
+	struct amqp_ssl_socket_t *self;
+	if (base->klass != &amqp_ssl_socket_class) {
+		amqp_abort("<%p> is not of type amqp_ssl_socket_t", base);
+	}
+	self = (struct amqp_ssl_socket_t *)base;
 	self->verify = verify;
 }
 
diff --git a/librabbitmq/amqp-socket.h b/librabbitmq/amqp-socket.h
index 1c21da0..a40d13e 100644
--- a/librabbitmq/amqp-socket.h
+++ b/librabbitmq/amqp-socket.h
@@ -37,7 +37,7 @@ typedef int (*amqp_socket_close_fn)(void *);
 typedef int (*amqp_socket_error_fn)(void *);
 typedef int (*amqp_socket_get_sockfd_fn)(void *);
 
-struct amqp_socket_t_ {
+struct amqp_socket_class_t {
 	amqp_socket_writev_fn writev;
 	amqp_socket_send_fn send;
 	amqp_socket_recv_fn recv;
@@ -47,6 +47,10 @@ struct amqp_socket_t_ {
 	amqp_socket_get_sockfd_fn get_sockfd;
 };
 
+struct amqp_socket_t_ {
+	const struct amqp_socket_class_t *klass;
+};
+
 ssize_t
 amqp_socket_writev(amqp_socket_t *self, const struct iovec *iov, int iovcnt);
 
diff --git a/librabbitmq/amqp-tcp-socket.c b/librabbitmq/amqp-tcp-socket.c
index 10430a9..2307c5c 100644
--- a/librabbitmq/amqp-tcp-socket.c
+++ b/librabbitmq/amqp-tcp-socket.c
@@ -29,7 +29,7 @@
 #include <stdlib.h>
 
 struct amqp_tcp_socket_t {
-	amqp_socket_t base;
+	const struct amqp_socket_class_t *klass;
 	int sockfd;
 };
 
@@ -117,6 +117,16 @@ amqp_tcp_socket_get_sockfd(void *base)
 	return self->sockfd;
 }
 
+static const struct amqp_socket_class_t amqp_tcp_socket_class = {
+	amqp_tcp_socket_writev, /* writev */
+	amqp_tcp_socket_send, /* send */
+	amqp_tcp_socket_recv, /* recv */
+	amqp_tcp_socket_open, /* open */
+	amqp_tcp_socket_close, /* close */
+	amqp_tcp_socket_error, /* error */
+	amqp_tcp_socket_get_sockfd /* get_sockfd */
+};
+
 amqp_socket_t *
 amqp_tcp_socket_new(void)
 {
@@ -124,13 +134,7 @@ amqp_tcp_socket_new(void)
 	if (!self) {
 		return NULL;
 	}
-	self->base.writev = amqp_tcp_socket_writev;
-	self->base.send = amqp_tcp_socket_send;
-	self->base.recv = amqp_tcp_socket_recv;
-	self->base.open = amqp_tcp_socket_open;
-	self->base.close = amqp_tcp_socket_close;
-	self->base.error = amqp_tcp_socket_error;
-	self->base.get_sockfd = amqp_tcp_socket_get_sockfd;
+	self->klass = &amqp_tcp_socket_class;
 	self->sockfd = -1;
 	return (amqp_socket_t *)self;
 }
@@ -138,6 +142,10 @@ amqp_tcp_socket_new(void)
 void
 amqp_tcp_socket_set_sockfd(amqp_socket_t *base, int sockfd)
 {
-	struct amqp_tcp_socket_t *self = (struct amqp_tcp_socket_t *)base;
+	struct amqp_tcp_socket_t *self;
+	if (base->klass != &amqp_tcp_socket_class) {
+		amqp_abort("<%p> is not of type amqp_tcp_socket_t", base);
+	}
+	self = (struct amqp_tcp_socket_t *)base;
 	self->sockfd = sockfd;
 }
diff --git a/librabbitmq/amqp_socket.c b/librabbitmq/amqp_socket.c
index d7be798..4ce2093 100644
--- a/librabbitmq/amqp_socket.c
+++ b/librabbitmq/amqp_socket.c
@@ -52,32 +52,32 @@
 ssize_t
 amqp_socket_writev(amqp_socket_t *self, const struct iovec *iov, int iovcnt)
 {
-  return self->writev(self, iov, iovcnt);
+  return self->klass->writev(self, iov, iovcnt);
 }
 
 ssize_t
 amqp_socket_send(amqp_socket_t *self, const void *buf, size_t len, int flags)
 {
-  return self->send(self, buf, len, flags);
+  return self->klass->send(self, buf, len, flags);
 }
 
 ssize_t
 amqp_socket_recv(amqp_socket_t *self, void *buf, size_t len, int flags)
 {
-  return self->recv(self, buf, len, flags);
+  return self->klass->recv(self, buf, len, flags);
 }
 
 int
 amqp_socket_open(amqp_socket_t *self, const char *host, int port)
 {
-  return self->open(self, host, port);
+  return self->klass->open(self, host, port);
 }
 
 int
 amqp_socket_close(amqp_socket_t *self)
 {
   if (self) {
-    return self->close(self);
+    return self->klass->close(self);
   }
   return 0;
 }
@@ -85,13 +85,13 @@ amqp_socket_close(amqp_socket_t *self)
 int
 amqp_socket_error(amqp_socket_t *self)
 {
-  return self->error(self);
+  return self->klass->error(self);
 }
 
 int
 amqp_socket_get_sockfd(amqp_socket_t *self)
 {
-  return self->get_sockfd(self);
+  return self->klass->get_sockfd(self);
 }
 
 int amqp_open_socket(char const *hostname,