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