diff --git a/README.md b/README.md
index 0085f47b..f1751b35 100644
--- a/README.md
+++ b/README.md
@@ -25,13 +25,16 @@ Both libraries cannot be installed at the same time. If you accidentally did ins
 * Modules:
   * Docker:
     - community.docker.docker_container: manage Docker containers
+    - community.docker.docker_container_exec: run commands in Docker containers
     - community.docker.docker_container_info: retrieve information on Docker containers
     - community.docker.docker_host_info: retrieve information on the Docker daemon
     - community.docker.docker_image: manage Docker images
     - community.docker.docker_image_info: retrieve information on Docker images
+    - community.docker.docker_image_load: load Docker images from archives
     - community.docker.docker_login: log in and out to/from registries
     - community.docker.docker_network: manage Docker networks
     - community.docker.docker_network_info: retrieve information on Docker networks
+    - community.docker.docker_plugin: manage Docker plugins
     - community.docker.docker_prune: prune Docker containers, images, networks, volumes, and build data
     - community.docker.docker_volume: manage Docker volumes
     - community.docker.docker_volume_info: retrieve information on Docker volumes
@@ -50,6 +53,8 @@ Both libraries cannot be installed at the same time. If you accidentally did ins
     - community.docker.docker_stack: manage Docker Stacks
     - community.docker.docker_stack_info: retrieve information on Docker Stacks
     - community.docker.docker_stack_task_info: retrieve information on tasks in Docker Stacks
+  * Other:
+    - current_container_facts: return facts about whether the module runs in a Docker container
 
 ## Using this collection
 
diff --git a/plugins/module_utils/common.py b/plugins/module_utils/common.py
index e364c3e2..5c3b79a3 100644
--- a/plugins/module_utils/common.py
+++ b/plugins/module_utils/common.py
@@ -606,7 +606,7 @@ class AnsibleDockerClientBase(Client):
 class AnsibleDockerClient(AnsibleDockerClientBase):
 
     def __init__(self, argument_spec=None, supports_check_mode=False, mutually_exclusive=None,
-                 required_together=None, required_if=None, min_docker_version=None,
+                 required_together=None, required_if=None, required_one_of=None, min_docker_version=None,
                  min_docker_api_version=None, option_minimal_versions=None,
                  option_minimal_versions_ignore_params=None, fail_results=None):
 
@@ -635,7 +635,9 @@ class AnsibleDockerClient(AnsibleDockerClientBase):
             supports_check_mode=supports_check_mode,
             mutually_exclusive=mutually_exclusive_params,
             required_together=required_together_params,
-            required_if=required_if)
+            required_if=required_if,
+            required_one_of=required_one_of,
+        )
 
         self.debug = self.module.params.get('debug')
         self.check_mode = self.module.check_mode
diff --git a/plugins/module_utils/socket_handler.py b/plugins/module_utils/socket_handler.py
new file mode 100644
index 00000000..36937ad7
--- /dev/null
+++ b/plugins/module_utils/socket_handler.py
@@ -0,0 +1,232 @@
+# Copyright (c) 2019-2021, Felix Fontein <felix@fontein.de>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+
+import os
+import os.path
+import socket as pysocket
+
+from ansible.module_utils.basic import missing_required_lib
+from ansible.module_utils.six import PY3
+
+try:
+    from docker.utils import socket as docker_socket
+    import struct
+except Exception:
+    # missing Docker SDK for Python handled in ansible_collections.community.docker.plugins.module_utils.common
+    pass
+
+from ansible_collections.community.docker.plugins.module_utils.socket_helper import (
+    make_unblocking,
+    shutdown_writing,
+    write_to_socket,
+)
+
+
+PARAMIKO_POLL_TIMEOUT = 0.01  # 10 milliseconds
+
+
+class DockerSocketHandlerBase(object):
+    def __init__(self, sock, selectors, log=None):
+        make_unblocking(sock)
+
+        self._selectors = selectors
+        if log is not None:
+            self._log = log
+        else:
+            self._log = lambda msg: True
+        self._paramiko_read_workaround = hasattr(sock, 'send_ready') and 'paramiko' in str(type(sock))
+
+        self._sock = sock
+        self._block_done_callback = None
+        self._block_buffer = []
+        self._eof = False
+        self._read_buffer = b''
+        self._write_buffer = b''
+        self._end_of_writing = False
+
+        self._current_stream = None
+        self._current_missing = 0
+        self._current_buffer = b''
+
+        self._selector = self._selectors.DefaultSelector()
+        self._selector.register(self._sock, self._selectors.EVENT_READ)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type, value, tb):
+        self._selector.close()
+
+    def set_block_done_callback(self, block_done_callback):
+        self._block_done_callback = block_done_callback
+        if self._block_done_callback is not None:
+            while self._block_buffer:
+                elt = self._block_buffer.remove(0)
+                self._block_done_callback(*elt)
+
+    def _add_block(self, stream_id, data):
+        if self._block_done_callback is not None:
+            self._block_done_callback(stream_id, data)
+        else:
+            self._block_buffer.append((stream_id, data))
+
+    def _read(self):
+        if self._eof:
+            return
+        if hasattr(self._sock, 'recv'):
+            try:
+                data = self._sock.recv(262144)
+            except Exception as e:
+                # After calling self._sock.shutdown(), OpenSSL's/urllib3's
+                # WrappedSocket seems to eventually raise ZeroReturnError in
+                # case of EOF
+                if 'OpenSSL.SSL.ZeroReturnError' in str(type(e)):
+                    self._eof = True
+                    return
+                else:
+                    raise
+        elif PY3 and isinstance(self._sock, getattr(pysocket, 'SocketIO')):
+            data = self._sock.read()
+        else:
+            data = os.read(self._sock.fileno())
+        if data is None:
+            # no data available
+            return
+        self._log('read {0} bytes'.format(len(data)))
+        if len(data) == 0:
+            # Stream EOF
+            self._eof = True
+            return
+        self._read_buffer += data
+        while len(self._read_buffer) > 0:
+            if self._current_missing > 0:
+                n = min(len(self._read_buffer), self._current_missing)
+                self._current_buffer += self._read_buffer[:n]
+                self._read_buffer = self._read_buffer[n:]
+                self._current_missing -= n
+                if self._current_missing == 0:
+                    self._add_block(self._current_stream, self._current_buffer)
+                    self._current_buffer = b''
+            if len(self._read_buffer) < 8:
+                break
+            self._current_stream, self._current_missing = struct.unpack('>BxxxL', self._read_buffer[:8])
+            self._read_buffer = self._read_buffer[8:]
+            if self._current_missing < 0:
+                # Stream EOF (as reported by docker daemon)
+                self._eof = True
+                break
+
+    def _handle_end_of_writing(self):
+        if self._end_of_writing and len(self._write_buffer) == 0:
+            self._end_of_writing = False
+            self._log('Shutting socket down for writing')
+            shutdown_writing(self._sock, self._log)
+
+    def _write(self):
+        if len(self._write_buffer) > 0:
+            written = write_to_socket(self._sock, self._write_buffer)
+            self._write_buffer = self._write_buffer[written:]
+            self._log('wrote {0} bytes, {1} are left'.format(written, len(self._write_buffer)))
+            if len(self._write_buffer) > 0:
+                self._selector.modify(self._sock, self._selectors.EVENT_READ | self._selectors.EVENT_WRITE)
+            else:
+                self._selector.modify(self._sock, self._selectors.EVENT_READ)
+            self._handle_end_of_writing()
+
+    def select(self, timeout=None, _internal_recursion=False):
+        if not _internal_recursion and self._paramiko_read_workaround and len(self._write_buffer) > 0:
+            # When the SSH transport is used, docker-py internally uses Paramiko, whose
+            # Channel object supports select(), but only for reading
+            # (https://github.com/paramiko/paramiko/issues/695).
+            if self._sock.send_ready():
+                self._write()
+                return True
+            while timeout is None or timeout > PARAMIKO_POLL_TIMEOUT:
+                result = self.select(PARAMIKO_POLL_TIMEOUT, _internal_recursion=True)
+                if self._sock.send_ready():
+                    self._read()
+                    result += 1
+                if result > 0:
+                    return True
+                if timeout is not None:
+                    timeout -= PARAMIKO_POLL_TIMEOUT
+        self._log('select... ({0})'.format(timeout))
+        events = self._selector.select(timeout)
+        for key, event in events:
+            if key.fileobj == self._sock:
+                self._log(
+                    'select event read:{0} write:{1}'.format(
+                        event & self._selectors.EVENT_READ != 0,
+                        event & self._selectors.EVENT_WRITE != 0))
+                if event & self._selectors.EVENT_READ != 0:
+                    self._read()
+                if event & self._selectors.EVENT_WRITE != 0:
+                    self._write()
+        result = len(events)
+        if self._paramiko_read_workaround and len(self._write_buffer) > 0:
+            if self._sock.send_ready():
+                self._write()
+                result += 1
+        return result > 0
+
+    def is_eof(self):
+        return self._eof
+
+    def end_of_writing(self):
+        self._end_of_writing = True
+        self._handle_end_of_writing()
+
+    def consume(self):
+        stdout = []
+        stderr = []
+
+        def append_block(stream_id, data):
+            if stream_id == docker_socket.STDOUT:
+                stdout.append(data)
+            elif stream_id == docker_socket.STDERR:
+                stderr.append(data)
+            else:
+                raise ValueError('{0} is not a valid stream ID'.format(stream_id))
+
+        self.end_of_writing()
+
+        self.set_block_done_callback(append_block)
+        while not self._eof:
+            self.select()
+        return b''.join(stdout), b''.join(stderr)
+
+    def write(self, str):
+        self._write_buffer += str
+        if len(self._write_buffer) == len(str):
+            self._write()
+
+
+class DockerSocketHandlerModule(DockerSocketHandlerBase):
+    def __init__(self, sock, module, selectors):
+        super(DockerSocketHandlerModule, self).__init__(sock, selectors, module.debug)
+
+
+def find_selectors(module):
+    try:
+        # ansible-base 2.10+ has selectors a compat version of selectors, which a bundled fallback:
+        from ansible.module_utils.compat import selectors
+        return selectors
+    except ImportError:
+        pass
+    try:
+        # Python 3.4+
+        import selectors
+        return selectors
+    except ImportError:
+        pass
+    try:
+        # backport package installed in the system
+        import selectors2
+        return selectors2
+    except ImportError:
+        pass
+    module.fail_json(msg=missing_required_lib('selectors2', reason='for handling stdin'))
diff --git a/plugins/module_utils/socket_helper.py b/plugins/module_utils/socket_helper.py
new file mode 100644
index 00000000..06d93446
--- /dev/null
+++ b/plugins/module_utils/socket_helper.py
@@ -0,0 +1,53 @@
+# Copyright (c) 2019-2021, Felix Fontein <felix@fontein.de>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+
+import fcntl
+import os
+import os.path
+import socket as pysocket
+
+from ansible.module_utils.six import PY3
+
+
+def make_unblocking(sock):
+    if hasattr(sock, '_sock'):
+        sock._sock.setblocking(0)
+    elif hasattr(sock, 'setblocking'):
+        sock.setblocking(0)
+    else:
+        fcntl.fcntl(sock.fileno(), fcntl.F_SETFL, fcntl.fcntl(sock.fileno(), fcntl.F_GETFL) | os.O_NONBLOCK)
+
+
+def _empty_writer(msg):
+    pass
+
+
+def shutdown_writing(sock, log=_empty_writer):
+    if hasattr(sock, 'shutdown_write'):
+        sock.shutdown_write()
+    elif hasattr(sock, 'shutdown'):
+        try:
+            sock.shutdown(pysocket.SHUT_WR)
+        except TypeError as e:
+            # probably: "TypeError: shutdown() takes 1 positional argument but 2 were given"
+            log('Shutting down for writing not possible; trying shutdown instead: {0}'.format(e))
+            sock.shutdown()
+    elif PY3 and isinstance(sock, getattr(pysocket, 'SocketIO')):
+        sock._sock.shutdown(pysocket.SHUT_WR)
+    else:
+        log('No idea how to signal end of writing')
+
+
+def write_to_socket(sock, data):
+    if hasattr(sock, '_send_until_done'):
+        # WrappedSocket (urllib3/contrib/pyopenssl) doesn't have `send`, but
+        # only `sendall`, which uses `_send_until_done` under the hood.
+        return sock._send_until_done(data)
+    elif hasattr(sock, 'send'):
+        return sock.send(data)
+    else:
+        return os.write(sock.fileno(), data)
diff --git a/plugins/modules/docker_container_exec.py b/plugins/modules/docker_container_exec.py
new file mode 100644
index 00000000..27532d8f
--- /dev/null
+++ b/plugins/modules/docker_container_exec.py
@@ -0,0 +1,256 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2021, Felix Fontein <felix@fontein.de>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = '''
+---
+module: docker_container_exec
+
+short_description: Execute command in a docker container
+
+version_added: 1.5.0
+
+description:
+  - Executes a command in a Docker container.
+
+options:
+  container:
+    type: str
+    required: true
+    description:
+      - The name of the container to execute the command in.
+  argv:
+    type: list
+    elements: str
+    description:
+      - The command to execute.
+      - Since this is a list of arguments, no quoting is needed.
+      - Exactly one of I(argv) and I(command) must be specified.
+  command:
+    type: str
+    description:
+      - The command to execute.
+      - Exactly one of I(argv) and I(command) must be specified.
+  chdir:
+    type: str
+    description:
+      - The directory to run the command in.
+  user:
+    type: str
+    description:
+      - If specified, the user to execute this command with.
+  stdin:
+    type: str
+    description:
+      - Set the stdin of the command directly to the specified value.
+  stdin_add_newline:
+    type: bool
+    default: true
+    description:
+      - If set to C(true), appends a newline to I(stdin).
+  strip_empty_ends:
+    type: bool
+    default: true
+    description:
+      - Strip empty lines from the end of stdout/stderr in result.
+  tty:
+    type: bool
+    default: false
+    description:
+      - Whether to allocate a TTY.
+
+extends_documentation_fragment:
+  - community.docker.docker
+  - community.docker.docker.docker_py_1_documentation
+notes:
+  - Does not support C(check_mode).
+author:
+  - "Felix Fontein (@felixfontein)"
+
+requirements:
+  - "L(Docker SDK for Python,https://docker-py.readthedocs.io/en/stable/) >= 1.8.0 (use L(docker-py,https://pypi.org/project/docker-py/) for Python 2.6)"
+  - "Docker API >= 1.20"
+'''
+
+EXAMPLES = '''
+- name: Run a simple command (command)
+  community.docker.docker_container_exec:
+    container: foo
+    command: /bin/bash -c "ls -lah"
+    chdir: /root
+  register: result
+
+- name: Print stdout
+  debug:
+    var: result.stdout
+
+- name: Run a simple command (argv)
+  community.docker.docker_container_exec:
+    container: foo
+    argv:
+      - /bin/bash
+      - "-c"
+      - "ls -lah > /dev/stderr"
+    chdir: /root
+  register: result
+
+- name: Print stderr lines
+  debug:
+    var: result.stderr_lines
+'''
+
+RETURN = '''
+stdout:
+    type: str
+    returned: success
+    description:
+      - The standard output of the container command.
+stderr:
+    type: str
+    returned: success
+    description:
+      - The standard error output of the container command.
+rc:
+    type: int
+    returned: success
+    sample: 0
+    description:
+      - The exit code of the command.
+'''
+
+import shlex
+import traceback
+
+from ansible.module_utils._text import to_text, to_bytes
+
+from ansible_collections.community.docker.plugins.module_utils.common import (
+    AnsibleDockerClient,
+    RequestException,
+)
+
+from ansible_collections.community.docker.plugins.module_utils.socket_helper import (
+    shutdown_writing,
+    write_to_socket,
+)
+
+from ansible_collections.community.docker.plugins.module_utils.socket_handler import (
+    find_selectors,
+    DockerSocketHandlerModule,
+)
+
+try:
+    from docker.errors import DockerException, APIError, NotFound
+except Exception:
+    # missing Docker SDK for Python handled in ansible.module_utils.docker.common
+    pass
+
+
+def main():
+    argument_spec = dict(
+        container=dict(type='str', required=True),
+        argv=dict(type='list', elements='str'),
+        command=dict(type='str'),
+        chdir=dict(type='str'),
+        user=dict(type='str'),
+        stdin=dict(type='str'),
+        stdin_add_newline=dict(type='bool', default=True),
+        strip_empty_ends=dict(type='bool', default=True),
+        tty=dict(type='bool', default=False),
+    )
+
+    client = AnsibleDockerClient(
+        argument_spec=argument_spec,
+        min_docker_api_version='1.20',
+        mutually_exclusive=[('argv', 'command')],
+        required_one_of=[('argv', 'command')],
+    )
+
+    container = client.module.params['container']
+    argv = client.module.params['argv']
+    command = client.module.params['command']
+    chdir = client.module.params['chdir']
+    user = client.module.params['user']
+    stdin = client.module.params['stdin']
+    strip_empty_ends = client.module.params['strip_empty_ends']
+    tty = client.module.params['tty']
+
+    if command is not None:
+        argv = shlex.split(command)
+
+    if stdin is not None and client.module.params['stdin_add_newline']:
+        stdin += '\n'
+
+    selectors = None
+    if stdin:
+        selectors = find_selectors(client.module)
+
+    try:
+        exec_data = client.exec_create(
+            container,
+            argv,
+            stdout=True,
+            stderr=True,
+            stdin=bool(stdin),
+            user=user or '',
+            workdir=chdir,
+        )
+        exec_id = exec_data['Id']
+
+        if selectors:
+            exec_socket = client.exec_start(
+                exec_id,
+                tty=tty,
+                detach=False,
+                socket=True,
+            )
+            try:
+                with DockerSocketHandlerModule(exec_socket, client.module, selectors) as exec_socket_handler:
+                    if stdin:
+                        exec_socket_handler.write(to_bytes(stdin))
+
+                    stdout, stderr = exec_socket_handler.consume()
+            finally:
+                exec_socket.close()
+        else:
+            stdout, stderr = client.exec_start(
+                exec_id,
+                tty=tty,
+                detach=False,
+                stream=False,
+                socket=False,
+                demux=True,
+            )
+
+        result = client.exec_inspect(exec_id)
+
+        stdout = to_text(stdout or b'')
+        stderr = to_text(stderr or b'')
+        if strip_empty_ends:
+            stdout = stdout.rstrip('\r\n')
+            stderr = stderr.rstrip('\r\n')
+
+        client.module.exit_json(
+            changed=True,
+            stdout=stdout,
+            stderr=stderr,
+            rc=result.get('ExitCode') or 0,
+        )
+    except NotFound:
+        client.fail('Could not find container "{0}"'.format(container))
+    except APIError as e:
+        if e.response and e.response.status_code == 409:
+            client.fail('The container "{0}" has been paused ({1})'.format(container, e))
+        client.fail('An unexpected docker error occurred: {0}'.format(e), exception=traceback.format_exc())
+    except DockerException as e:
+        client.fail('An unexpected docker error occurred: {0}'.format(e), exception=traceback.format_exc())
+    except RequestException as e:
+        client.fail('An unexpected requests error occurred when docker-py tried to talk to the docker daemon: {0}'.format(e), exception=traceback.format_exc())
+
+
+if __name__ == '__main__':
+    main()
diff --git a/plugins/plugin_utils/socket_handler.py b/plugins/plugin_utils/socket_handler.py
index bbaa1565..4bc667be 100644
--- a/plugins/plugin_utils/socket_handler.py
+++ b/plugins/plugin_utils/socket_handler.py
@@ -5,217 +5,13 @@ from __future__ import (absolute_import, division, print_function)
 __metaclass__ = type
 
 
-import fcntl
-import os
-import os.path
-import socket as pysocket
-
 from ansible.compat import selectors
-from ansible.module_utils.six import PY3
 
-try:
-    from docker.utils import socket as docker_socket
-    import struct
-except Exception:
-    # missing Docker SDK for Python handled in ansible_collections.community.docker.plugins.module_utils.common
-    pass
+from ansible_collections.community.docker.plugins.module_utils.socket_handler import (
+    DockerSocketHandlerBase,
+)
 
 
-PARAMIKO_POLL_TIMEOUT = 0.01  # 10 milliseconds
-
-
-class DockerSocketHandler:
-    def __init__(self, display, sock, container=None):
-        if hasattr(sock, '_sock'):
-            sock._sock.setblocking(0)
-        elif hasattr(sock, 'setblocking'):
-            sock.setblocking(0)
-        else:
-            fcntl.fcntl(sock.fileno(), fcntl.F_SETFL, fcntl.fcntl(sock.fileno(), fcntl.F_GETFL) | os.O_NONBLOCK)
-
-        self._display = display
-        self._paramiko_read_workaround = hasattr(sock, 'send_ready') and 'paramiko' in str(type(sock))
-
-        self._container = container
-
-        self._sock = sock
-        self._block_done_callback = None
-        self._block_buffer = []
-        self._eof = False
-        self._read_buffer = b''
-        self._write_buffer = b''
-        self._end_of_writing = False
-
-        self._current_stream = None
-        self._current_missing = 0
-        self._current_buffer = b''
-
-        self._selector = selectors.DefaultSelector()
-        self._selector.register(self._sock, selectors.EVENT_READ)
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, type, value, tb):
-        self._selector.close()
-
-    def set_block_done_callback(self, block_done_callback):
-        self._block_done_callback = block_done_callback
-        if self._block_done_callback is not None:
-            while self._block_buffer:
-                elt = self._block_buffer.remove(0)
-                self._block_done_callback(*elt)
-
-    def _add_block(self, stream_id, data):
-        if self._block_done_callback is not None:
-            self._block_done_callback(stream_id, data)
-        else:
-            self._block_buffer.append((stream_id, data))
-
-    def _read(self):
-        if self._eof:
-            return
-        if hasattr(self._sock, 'recv'):
-            try:
-                data = self._sock.recv(262144)
-            except Exception as e:
-                # After calling self._sock.shutdown(), OpenSSL's/urllib3's
-                # WrappedSocket seems to eventually raise ZeroReturnError in
-                # case of EOF
-                if 'OpenSSL.SSL.ZeroReturnError' in str(type(e)):
-                    self._eof = True
-                    return
-                else:
-                    raise
-        elif PY3 and isinstance(self._sock, getattr(pysocket, 'SocketIO')):
-            data = self._sock.read()
-        else:
-            data = os.read(self._sock.fileno())
-        if data is None:
-            # no data available
-            return
-        self._display.vvvv('read {0} bytes'.format(len(data)), host=self._container)
-        if len(data) == 0:
-            # Stream EOF
-            self._eof = True
-            return
-        self._read_buffer += data
-        while len(self._read_buffer) > 0:
-            if self._current_missing > 0:
-                n = min(len(self._read_buffer), self._current_missing)
-                self._current_buffer += self._read_buffer[:n]
-                self._read_buffer = self._read_buffer[n:]
-                self._current_missing -= n
-                if self._current_missing == 0:
-                    self._add_block(self._current_stream, self._current_buffer)
-                    self._current_buffer = b''
-            if len(self._read_buffer) < 8:
-                break
-            self._current_stream, self._current_missing = struct.unpack('>BxxxL', self._read_buffer[:8])
-            self._read_buffer = self._read_buffer[8:]
-            if self._current_missing < 0:
-                # Stream EOF (as reported by docker daemon)
-                self._eof = True
-                break
-
-    def _handle_end_of_writing(self):
-        if self._end_of_writing and len(self._write_buffer) == 0:
-            self._end_of_writing = False
-            self._display.vvvv('Shutting socket down for writing', host=self._container)
-            if hasattr(self._sock, 'shutdown_write'):
-                self._sock.shutdown_write()
-            elif hasattr(self._sock, 'shutdown'):
-                try:
-                    self._sock.shutdown(pysocket.SHUT_WR)
-                except TypeError as e:
-                    # probably: "TypeError: shutdown() takes 1 positional argument but 2 were given"
-                    self._display.vvvv('Shutting down for writing not possible; trying shutdown instead: {0}'.format(e), host=self._container)
-                    self._sock.shutdown()
-            elif PY3 and isinstance(self._sock, getattr(pysocket, 'SocketIO')):
-                self._sock._sock.shutdown(pysocket.SHUT_WR)
-            else:
-                self._display.vvvv('No idea how to signal end of writing', host=self._container)
-
-    def _write(self):
-        if len(self._write_buffer) > 0:
-            if hasattr(self._sock, '_send_until_done'):
-                # WrappedSocket (urllib3/contrib/pyopenssl) doesn't have `send`, but
-                # only `sendall`, which uses `_send_until_done` under the hood.
-                written = self._sock._send_until_done(self._write_buffer)
-            elif hasattr(self._sock, 'send'):
-                written = self._sock.send(self._write_buffer)
-            else:
-                written = os.write(self._sock.fileno(), self._write_buffer)
-            self._write_buffer = self._write_buffer[written:]
-            self._display.vvvv('wrote {0} bytes, {1} are left'.format(written, len(self._write_buffer)), host=self._container)
-            if len(self._write_buffer) > 0:
-                self._selector.modify(self._sock, selectors.EVENT_READ | selectors.EVENT_WRITE)
-            else:
-                self._selector.modify(self._sock, selectors.EVENT_READ)
-            self._handle_end_of_writing()
-
-    def select(self, timeout=None, _internal_recursion=False):
-        if not _internal_recursion and self._paramiko_read_workaround and len(self._write_buffer) > 0:
-            # When the SSH transport is used, docker-py internally uses Paramiko, whose
-            # Channel object supports select(), but only for reading
-            # (https://github.com/paramiko/paramiko/issues/695).
-            if self._sock.send_ready():
-                self._write()
-                return True
-            while timeout is None or timeout > PARAMIKO_POLL_TIMEOUT:
-                result = self.select(PARAMIKO_POLL_TIMEOUT, _internal_recursion=True)
-                if self._sock.send_ready():
-                    self._read()
-                    result += 1
-                if result > 0:
-                    return True
-                if timeout is not None:
-                    timeout -= PARAMIKO_POLL_TIMEOUT
-        self._display.vvvv('select... ({0})'.format(timeout), host=self._container)
-        events = self._selector.select(timeout)
-        for key, event in events:
-            if key.fileobj == self._sock:
-                self._display.vvvv(
-                    'select event read:{0} write:{1}'.format(event & selectors.EVENT_READ != 0, event & selectors.EVENT_WRITE != 0),
-                    host=self._container)
-                if event & selectors.EVENT_READ != 0:
-                    self._read()
-                if event & selectors.EVENT_WRITE != 0:
-                    self._write()
-        result = len(events)
-        if self._paramiko_read_workaround and len(self._write_buffer) > 0:
-            if self._sock.send_ready():
-                self._write()
-                result += 1
-        return result > 0
-
-    def is_eof(self):
-        return self._eof
-
-    def end_of_writing(self):
-        self._end_of_writing = True
-        self._handle_end_of_writing()
-
-    def consume(self):
-        stdout = []
-        stderr = []
-
-        def append_block(stream_id, data):
-            if stream_id == docker_socket.STDOUT:
-                stdout.append(data)
-            elif stream_id == docker_socket.STDERR:
-                stderr.append(data)
-            else:
-                raise ValueError('{0} is not a valid stream ID'.format(stream_id))
-
-        self.end_of_writing()
-
-        self.set_block_done_callback(append_block)
-        while not self._eof:
-            self.select()
-        return b''.join(stdout), b''.join(stderr)
-
-    def write(self, str):
-        self._write_buffer += str
-        if len(self._write_buffer) == len(str):
-            self._write()
+class DockerSocketHandler(DockerSocketHandlerBase):
+    def __init__(self, display, sock, log=None, container=None):
+        super(DockerSocketHandler, self).__init__(sock, selectors, log=lambda msg: display.vvvv(msg, host=container))
diff --git a/tests/integration/targets/docker_container_exec/aliases b/tests/integration/targets/docker_container_exec/aliases
new file mode 100644
index 00000000..02b78723
--- /dev/null
+++ b/tests/integration/targets/docker_container_exec/aliases
@@ -0,0 +1,2 @@
+shippable/posix/group4
+destructive
diff --git a/tests/integration/targets/docker_container_exec/meta/main.yml b/tests/integration/targets/docker_container_exec/meta/main.yml
new file mode 100644
index 00000000..07da8c6d
--- /dev/null
+++ b/tests/integration/targets/docker_container_exec/meta/main.yml
@@ -0,0 +1,3 @@
+---
+dependencies:
+  - setup_docker
diff --git a/tests/integration/targets/docker_container_exec/tasks/main.yml b/tests/integration/targets/docker_container_exec/tasks/main.yml
new file mode 100644
index 00000000..10e26980
--- /dev/null
+++ b/tests/integration/targets/docker_container_exec/tasks/main.yml
@@ -0,0 +1,181 @@
+---
+####################################################################
+# WARNING: These are designed specifically for Ansible tests       #
+# and should not be used as examples of how to write Ansible roles #
+####################################################################
+
+- block:
+  - name: Create random container name
+    set_fact:
+      cname: "{{ 'ansible-test-%0x' % ((2**32) | random) }}"
+
+  - name: Make sure container is not there
+    docker_container:
+      name: "{{ cname }}"
+      state: absent
+      force_kill: yes
+
+  - name: Execute in a non-present container
+    docker_container_exec:
+      container: "{{ cname }}"
+      command: "/bin/bash -c 'ls -a'"
+    register: result
+    ignore_errors: true
+
+  - assert:
+      that:
+        - result is failed
+        - "'Could not find container' in result.msg"
+
+  - name: Make sure container exists
+    docker_container:
+      name: "{{ cname }}"
+      image: "{{ docker_test_image_alpine }}"
+      command: '/bin/sh -c "sleep 10m"'
+      state: started
+      force_kill: yes
+
+  - name: Execute in a present container (command)
+    docker_container_exec:
+      container: "{{ cname }}"
+      command: "/bin/sh -c 'ls -a'"
+    register: result_cmd
+
+  - assert:
+      that:
+        - result_cmd.rc == 0
+        - "'stdout' in result_cmd"
+        - "'stdout_lines' in result_cmd"
+        - "'stderr' in result_cmd"
+        - "'stderr_lines' in result_cmd"
+
+  - name: Execute in a present container (argv)
+    docker_container_exec:
+      container: "{{ cname }}"
+      argv:
+        - /bin/sh
+        - '-c'
+        - ls -a
+    register: result_argv
+
+  - assert:
+      that:
+        - result_argv.rc == 0
+        - "'stdout' in result_argv"
+        - "'stdout_lines' in result_argv"
+        - "'stderr' in result_argv"
+        - "'stderr_lines' in result_argv"
+        - result_cmd.stdout == result_argv.stdout
+
+  - name: Execute in a present container (cat without stdin)
+    docker_container_exec:
+      container: "{{ cname }}"
+      argv:
+        - /bin/sh
+        - '-c'
+        - cat
+    register: result
+
+  - assert:
+      that:
+        - result.rc == 0
+        - result.stdout == ''
+        - result.stdout_lines == []
+        - result.stderr == ''
+        - result.stderr_lines == []
+
+  - name: Execute in a present container (cat with stdin)
+    docker_container_exec:
+      container: "{{ cname }}"
+      argv:
+        - /bin/sh
+        - '-c'
+        - cat
+      stdin: Hello world!
+      strip_empty_ends: false
+    register: result
+
+  - assert:
+      that:
+        - result.rc == 0
+        - result.stdout == 'Hello world!\n'
+        - result.stdout_lines == ['Hello world!']
+        - result.stderr == ''
+        - result.stderr_lines == []
+
+  - name: Execute in a present container (cat with stdin, no newline)
+    docker_container_exec:
+      container: "{{ cname }}"
+      argv:
+        - /bin/sh
+        - '-c'
+        - cat
+      stdin: Hello world!
+      stdin_add_newline: false
+      strip_empty_ends: false
+    register: result
+
+  - assert:
+      that:
+        - result.rc == 0
+        - result.stdout == 'Hello world!'
+        - result.stdout_lines == ['Hello world!']
+        - result.stderr == ''
+        - result.stderr_lines == []
+
+  - name: Execute in a present container (cat with stdin, newline but stripping)
+    docker_container_exec:
+      container: "{{ cname }}"
+      argv:
+        - /bin/sh
+        - '-c'
+        - cat
+      stdin: Hello world!
+      stdin_add_newline: true
+      strip_empty_ends: true
+    register: result
+
+  - assert:
+      that:
+        - result.rc == 0
+        - result.stdout == 'Hello world!'
+        - result.stdout_lines == ['Hello world!']
+        - result.stderr == ''
+        - result.stderr_lines == []
+
+  - name: Prepare long string
+    set_fact:
+      very_long_string: "{{ 'something long ' * 10000 }}"
+      very_long_string2: "{{ 'something else ' * 5000 }}"
+
+  - name: Execute in a present container (long stdin)
+    docker_container_exec:
+      container: "{{ cname }}"
+      argv:
+        - /bin/sh
+        - '-c'
+        - cat
+      stdin: |-
+        {{ very_long_string }}
+        {{ very_long_string2 }}
+    register: result
+
+  - assert:
+      that:
+        - result.rc == 0
+        - result.stdout == very_long_string ~ '\n' ~ very_long_string2
+        - result.stdout_lines == [very_long_string, very_long_string2]
+        - result.stderr == ''
+        - result.stderr_lines == []
+
+  always:
+  - name: Cleanup
+    docker_container:
+      name: "{{ cname }}"
+      state: absent
+      force_kill: yes
+
+  when: docker_py_version is version('1.8.0', '>=') and docker_api_version is version('1.20', '>=')
+
+- fail: msg="Too old docker / docker-py version to run docker_container_exec tests!"
+  when: not(docker_py_version is version('1.8.0', '>=') and docker_api_version is version('1.20', '>=')) and (ansible_distribution != 'CentOS' or ansible_distribution_major_version|int > 6)