(self, address)
| 881 | self.createSocket() |
| 882 | |
| 883 | def _connect_unixsocket(self, address): |
| 884 | use_socktype = self.socktype |
| 885 | if use_socktype is None: |
| 886 | use_socktype = socket.SOCK_DGRAM |
| 887 | self.socket = socket.socket(socket.AF_UNIX, use_socktype) |
| 888 | try: |
| 889 | self.socket.connect(address) |
| 890 | # it worked, so set self.socktype to the used type |
| 891 | self.socktype = use_socktype |
| 892 | except OSError: |
| 893 | self.socket.close() |
| 894 | if self.socktype is not None: |
| 895 | # user didn't specify falling back, so fail |
| 896 | raise |
| 897 | use_socktype = socket.SOCK_STREAM |
| 898 | self.socket = socket.socket(socket.AF_UNIX, use_socktype) |
| 899 | try: |
| 900 | self.socket.connect(address) |
| 901 | # it worked, so set self.socktype to the used type |
| 902 | self.socktype = use_socktype |
| 903 | except OSError: |
| 904 | self.socket.close() |
| 905 | raise |
| 906 | |
| 907 | def createSocket(self): |
| 908 | """ |
no test coverage detected