1번 요구조건 분석(문제 정의)

1. 아래 요구사항을 만족하는 시스템을 구현하시오. - Broker와 일반 노드로 구성

- 일반 노드는 아래와 같이 실행 시 자신이 publish할 topic과 subscribe할 topic을

broker에게 알려줌

예: node1 –p “topic1” -s “topic2”

- 일반 노드는 하나의 topic만 publish 할 수 있음

- 일반 노드는 하나의 topic만 subscribe할 수 있음

- 일반 노드는 publish 또는 subscribe 중 하나만 수행할 수도 있음

- Broker는 일반 노드가 실행되면 topic 이름을 확인해 한 topic을 publish하는 노드와

subscribe하는 노드가 서로의 주소를 알 수 있도록 함

- Broker는 각 일반 노드가 어떤 topic을 publish하고 subscribe하는지 관리함

- Broker는 최대 5개의 일반 노드를 관리할 수 있음

- 일반 노드들은 broker를 통해 서로의 주소가 확인되면 TCP 연결을 통해 서로를

연결함

1번 요구사항 알고리즘(문제 분석)

가장 처음에는 클라이언트가 브로커에게 최대 4개의 argument를 string으로 변환 후 보낼 수 있도록 하여 브로커가 string을 확인 후 각 argument에 어떤 값들이 들어왔는지 확인하는 방식을 이용한다.

위의 그림처럼 -p name인지, -s name인지, -p name -s name인지 파악하여 각 flag로 나누고

해당하는 flag의 조건문에서 Client가 어떤 방식을 취할지 Broker가 설정해준다.

flag가 1일 때(Publisher)

Broker 입장

Client이 가지는 제한사항이 없으니 보내주고, Broker Table에 업데이트한다.

Client 입장

브로커에서 보내준 size값을 이용하여(15+‘p’라는 값을 암호 키처럼 보내준다.) 자신의 아이디와 포트 번호를 알도록 해주고 TCP를 이용하여 bind, listen을 통해 accept를 할 준비를 한다. 이때 accept과정은 Broker이 Subscriber이 들어온 경우 while문에서 ‘p’라는 암호 키를 보내주면 accept를 준비하도록 설계되어있다. accept를 성공하면 mutex_lock을 통해 스레드를 생성하고 mutex_unlock를 해준다. 이때 스레드(threadfunc)에서는 sub에게 메시지를 받게 되면 "Hello subscriber~"를 통해 subscriber과 통신하고 있음을 확인할 수 있다.

flag가 2일 때(Subscriber)

Broker 입장

Subscriber일 때는 Publisher가 존재하는지 확인해야한다.(Name matching)

따라서 이름을 비교하는 알고리즘을 이용한다. 이 알고리즘은 for문으로 strcmp를 이용하여 Broker Table에서 찾아내는 방식을 이용한다.

방이 존재하지 않는다면 flag가 -2로 변하고, “ERROR :: Room is not exist”이라는 메시지를 클라이언트에 보내주며 클라이언트를 종료 시킨다.

방이 존재한다면 클라이언트에게 접속 시도를 할 수 있도록 해주고, Subscriber는 Publisher에게 접속을 시도하게 되고, Broker Table에 업데이트한다.

Client 입장

브로커에서 보내준 size값을 이용하여(20+'s'라는 값을 암호 키처럼 보내준다.) 자신의 아이디와 포트 번호를 알도록 해주고 connect를 준비한다. TCP를 통해 connect를 성공하면 while문을 통해 fgets에서 pub에게 보낼 말을 칠 수 있고, pub과 연결이 되었음을 확인하는 "Hello subscriber~"이라는 메시지를 수신하게 된다.

flag가 3일 때(Publisher and Subscriber)

Broker 입장

flag1과 flag2가 가지는 조건을 이용한다. 따라서 for문으로 strcmp를 통해 Broker Table에서 Publisher이 있는지 찾아내는 알고리즘을 동일하게 이용하고, Broker Table에 업데이트한다. 이때도 마찬가지로 방이 존재하지 않을 경우 flag가 -3이 되어 “ERROR :: Room is not exist”이라는 메시지를 클라이언트에 보내주며 클라이언트를 종료 시킨다.

Client 입장

브로커에서 보내준 size값을 이용하여(20+'z'라는 값을 암호 키처럼 보내준다.) 자신의 아이디와 포트 번호를 알도록 해주고 mutex_lock를 통해 threadfuncS라는 스레드를 생성한다. 이 스레드는 Subscriber 역할 및 Publisher 역할을 할 수 있는 스레드로써 TCP를 통해 Publisher 및 Subscriber와 연결을 한다.  그리고 while문을 통해 Publisher에게 메시를 보내고, 정상적으로 통신이 되었다면 Publisher에게 "Hello subscirber~"이라는 메시지를 받게 된다. 그리고 메인 스레드에서는 bind, listen을 통해 accept를 준비하고 마찬가지로 'p'라는 암호 키를 받으면 accept를 할 수 있도록 하여 publish와 subscribe를 동시에 할 수 있도록 하였다.

1번 소스 코드

<Test.h>

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15	//testh.h #include <pthread.h> #include <semaphore.h> #include <stdio.h> #include <string.h> #include <stdlib.h> #include <netinet/in.h> #include <sys/socket.h> #include <arpa/inet.h> #include <unistd.h> #include <signal.h> #define PORT 20162 #define BUFFER_SIZE 300 #define LISTEN_QUEUE_SIZE 5 #define TIMEOUT 5	cs

<Server.c>

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

//server.cpp

# include "testh.h"

// 뮤텍스 생성 및 컨디션 생성(UDP용)

pthread_mutex_t udpmutex = PTHREAD_MUTEX_INITIALIZER;

pthread_cond_t udpcond = PTHREAD_COND_INITIALIZER;

int total = 0;

int count = 0;

int empty[5];

int emptyPos = 0;

int emptyFlag = 0;

int errorFlag = 0;

int now;

char readBuff[BUFFER_SIZE];

char sendBuff[BUFFER_SIZE];

int listenFD;

int connectFD;

int target; // 몇 번 방인지 확인하는 변수

int receivedBytes;

struct sockaddr_in listenSocket, connectSocket;

int connect_addr_size = sizeof(connectSocket);

// 알람을 통해 테이블을 갱신한다

void alarmHandler(int sig)

{

count++;

if(count == TIMEOUT)

{

printf("Time Out\n");

exit(0);

}

// 받은 값이 없으면 alarm 시그널 호출

if(receivedBytes == 0)

{

alarm(1);

}

else

{

//응답을 받았으면 alarm 시그널을 취소.

alarm(0);

count = 0;

receivedBytes = 0;

}

}

// 테이블 생성

struct Table

{

char createroom[100];

char joinroom[100];

//int connectPort[5];

int pub;

int sub;

int id;

int fd;

int port;

};

struct Table table[5]; // 최대 5개 서버 가능

void sendport(int port)

{

// port를 char형으로 변환하여 전송하는 알고리즘

char bufftmp[100];

int portTmp = port;

int i;

for(i = 0; ;i ++)

{

if(portTmp / 10 != 0)

{

bufftmp[i] = (portTmp % 10) + 48; // ASCII

portTmp /= 10;

}

else

break;

}

bufftmp[i] = (portTmp % 10) + 48; // 마지막 값

bufftmp[i+1] = '\0';

int j = strlen(bufftmp);

for(i = 0; j > 0 ; i++)

{

readBuff[j-1] = bufftmp[i];

j--;

}

readBuff[i] = '\0';

// publisher에게는 subscriber의 port를 준다.

sendto(listenFD, readBuff, strlen(readBuff),0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

}

int getinit()

{

int i = 0;

int j = 0;

int flag = 0;

// 테이블에 빈자리가 있다면 빈자리를 채우는 알고리즘

if(emptyFlag == 1)

{

emptyFlag = 0;

emptyPos--;

}

else if(emptyPos != 0)

{

now = empty[emptyPos-1];

//emptyPos--;

emptyFlag = 1;

i = 5;

}

else if(emptyPos == 0 && emptyFlag == 0)

{

int tmpi;

for(tmpi = 0; tmpi < 5; tmpi++)

{

if(table[tmpi].fd == 0)

{

now = tmpi;

//errorFlag = 0;

break;

}

}

}

table[now].createroom[0] = '\0';

table[now].joinroom[0] = '\0';

table[now].pub = 0;

table[now].sub = 0;

table[now].id = now;

table[now].fd = 0;

table[now].port = 0;

int tmpi;

for(tmpi = 0; tmpi < 5; tmpi++)

{

if(table[tmpi].fd == 0)

{

errorFlag == 1 ? (now = tmpi) : (now = tmpi);

errorFlag = 0;

break;

}

}

// 시작이 -p or -s가 아닐 때

if(!((readBuff[0] == '-' && readBuff[1] == 's' && readBuff[2] == ' ') || (readBuff[0] == '-' && readBuff[1] == 'p' && readBuff[2] == ' ')))

{

sendto(listenFD, "ERROR :: Usage :: '-p name -s name' or '-p name' or '-s name'", 20+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket)+30);

return -1;

}

// 시작이 -p 네임 일 때

if(readBuff[0] == '-' && readBuff[1] == 'p' && readBuff[2] == ' ' && !strstr(readBuff,"-s"))

{

table[now].pub = 1;

table[now].fd = listenFD;

table[now].port = connectSocket.sin_port;

i = 3;

for(i; readBuff[i] != ' '; i++,j++)

table[now].createroom[j] = readBuff[i];

table[now].createroom[j] = '\0';

i++;

flag = 1;

}

j = 0;

// 시작이 -s 네임 일 때

if(readBuff[0] == '-' && readBuff[1] == 's' && readBuff[2] == ' ')

{

table[now].sub = 1;

table[now].fd = listenFD;

table[now].port = connectSocket.sin_port;

i += 3;

for(i; readBuff[i] != ' '; i++,j++)

table[now].joinroom[j] = readBuff[i];

table[now].joinroom[j] = '\0';

i++;

flag = 2;

}

// 시작이 -p 네임 -s 네임 일 때

//if(readBuff[i] == '-'  readBuff[i+1] == 'p'  readBuff[2] == ' ')

if(strstr(readBuff,"-p") && strstr(readBuff,"-s"))

{

table[now].pub = 1;

table[now].sub = 1;

table[now].fd = listenFD;

table[now].port = connectSocket.sin_port;

for(i = 3; readBuff[i] != ' '; i++,j++)

table[now].createroom[j] = readBuff[i];

table[now].createroom[j] = '\0';

i += 4;

j = 0;

for(i; i < strlen(readBuff); i++,j++)

{

table[now].joinroom[j] = readBuff[i];

}

table[now].joinroom[j] = '\0';

flag = 3;

i = 5;

}

// -p 네임 -'이상한 문자' 일 때

else if(readBuff[i] == '-' && readBuff[i+1] != 's' && readBuff[2] == ' ')

{

sendto(listenFD, "ERROR :: Usage :: '-p name -s name' or '-p name' or '-s name'", 20+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

return -1;

}

// 현재 들어온 값 보여준다.

printf("id :: %d\npub :: %d\nsub :: %d\nfd :: %2d\nport :: %2d\nname :: %s\nsname :: %s\n",table[now].id,table[now].pub,table[now].sub,table[i].fd, table[i].port,table[now].createroom,table[now].joinroom);

// 갱신된 테이블 값 보여준다.

printf("잘못된 입력 값도 받아 오는 [Table]\n");

for(i = 0; i < 5; i ++)

{

printf("id :: %d pub :: %d sub :: %d fd :: %2d port :: %5d pname :: %5s sname :: %5s\n",table[i].id,table[i].pub,table[i].sub,table[i].fd, table[i].port,table[i].createroom,table[i].joinroom);

}

// 플래그 상태(1 :: pub, 2:: sub, 3 :: pub  sub)

// 상태에 따라 pub, sub, pub&sub인지 알려준다.

printf("flag :: %d\n",flag);

if(flag == 1)

{

sendto(listenFD, "Broker Connect Completely", table[now].id + 15+'p', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

}

else if(flag == 2)

{

flag = -2;

for(i = 0; i < 5; i ++)

{

// 방이 생성되어있는지 확인한다.

if(strcmp(table[i].createroom,table[now].joinroom) == 0)

{

/*

알고리즘 ::

우선 sub에게 방이 존재하니 커넥트를 완료했음을 알린다.

그리고 pub에게 sub가 입장함을 알린다.

그 후, sub가 fd를 전달해주면 브로커는 sub의 fd를

pub에게 전달해주고, pub는 그것을 수신하게되면

pub의 fd를 다시 브로커에게 전달해준다.

이것을 sub에게 마지막으로 전달해주면 완료.

*/

sendto(listenFD, "Broker Connect Completely", table[now].id + 20+'s', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

sendport(i);

flag = 2;

//                int subFD = recvfrom(listenFD, readBuff, BUFFER_SIZE, 0, (struct sockaddr*)&connectSocket, &connect_addr_size);

int tmpPort = connectSocket.sin_port; // sub의 포트

// pub에게 sub가 입장함을 알려준다.

connectSocket.sin_port = table[i].port;

sendto(listenFD, "Try to accept", 'p', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

receivedBytes = recvfrom(listenFD, readBuff, BUFFER_SIZE, 0, (struct sockaddr*)&connectSocket, &connect_addr_size);

if(receivedBytes == (int)(50 + 'e'))

{

table[now].createroom[0] = '\0';

table[now].joinroom[0] = '\0';

table[now].pub = 0;

table[now].sub = 0;

table[now].id = now;

table[now].fd = 0;

table[now].port = 0;

//total --;

return -1;

}

break;

}

}

// 방이 없을경우 40 + 'e' 아스키 코드 전송

if(flag == -2)

{

errorFlag = 1;

sendto(listenFD, "ERROR :: Room is not exist", 40+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

table[now].createroom[0] = '\0';

table[now].joinroom[0] = '\0';

table[now].pub = 0;

table[now].sub = 0;

table[now].id = 0;

table[now].fd = 0;

table[now].port = 0;

return -1;

}

}

else if(flag == 3)

{

flag = -3;

for(i = 0; i < 5; i ++)

{

if(i != now && strcmp(table[i].createroom,table[now].joinroom) == 0)

{

sendto(listenFD, "Broker Connect Completely", table[now].id + 20+'z', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

// pub의 포트 가져온다

sendto(listenFD, "p",  table[i].id+1,0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

int tmpPort = connectSocket.sin_port; // sub의 포트

// pub에게 sub가 입장함을 알려준다.

connectSocket.sin_port = table[i].port;

sendto(listenFD, "Try to accept", 'p', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

flag = 3;

break;

}

}

// 방이 없을경우 40 + 'e' 아스키 코드 전송

if(flag == -3)

{

errorFlag = 1;

sendto(listenFD, "ERROR :: Room is not exist", 40+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

table[now].createroom[0] = '\0';

table[now].joinroom[0] = '\0';

table[now].pub = 0;

table[now].sub = 0;

table[now].id = 0;

table[now].fd = 0;

table[now].port = 0;

return -1;

}

}

return 1;

}

int main()

{

int i;

int flag = 0;

pthread_t threadID[5];

memset(&listenSocket, 0, sizeof(listenSocket));

memset(&listenSocket, 0, sizeof(connectSocket));

listenSocket.sin_family = AF_INET;

listenSocket.sin_addr.s_addr = htonl(INADDR_ANY);

listenSocket.sin_port = htons(PORT);

// 브로커 서버 소켓 생성(UDP :: SOCK_DGRAM)

if((listenFD = socket(AF_INET, SOCK_DGRAM, 0)) == -1)

{

printf("Server :: Can not Open Socket\n");

exit(0);

}

if(bind(listenFD, (struct sockaddr *) &listenSocket, sizeof(listenSocket)) == -1)

{

printf("Can not bind\n");

return -1;

}

printf("Waiting for clients...\n");

while(1)

{

receivedBytes = recvfrom(listenFD, readBuff, BUFFER_SIZE, 0, (struct sockaddr*)&connectSocket, &connect_addr_size);

readBuff[receivedBytes] = '\0';

printf("--------------------\n");

//printf("::: %s\n", readBuff);

// 종료 메시지를 보내온다면 테이블 정리를 한다.

if(0 + 'e' <= receivedBytes && receivedBytes <= 5 + 'e')

{

total--;

int tmpID = receivedBytes - 'e';

printf("지금 종료한 id :: %d\n",tmpID);

// 테이블의 빈자리를 알게 해준다.

empty[emptyPos] = tmpID;

emptyPos++;

table[tmpID].createroom[0] = '\0';

table[tmpID].joinroom[0] = '\0';

table[tmpID].pub = 0;

table[tmpID].sub = 0;

table[tmpID].id = 0;

table[tmpID].fd = 0;

table[tmpID].port = 0;

}

if(total < 5)

flag = getinit(); // -p 네임 -s 네임 확인

else

sendto(listenFD, "ERROR :: Client is full", 30+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

flag == 1 ? total++ : 0;

flag = 0;

// 무언가 갱신이 이상할 때

if(total < 0)

{

printf("[Warning] Total value is strange, Please excute again\n");

return 0;

}

printf("total :: %d\n",total);

// 갱신된 테이블 값 보여준다.

printf("오류를 처리한 최종 [Table]\n");

for(i = 0; i < 5; i ++)

{

printf("id :: %d pub :: %d sub :: %d fd :: %2d port :: %5d pname :: %5s sname :: %5s\n",table[i].id,table[i].pub,table[i].sub,table[i].fd, table[i].port,table[i].createroom,table[i].joinroom);

}

printf("\n");

fflush(stdout);

}

}

Colored by Color Scripter

cs

<Client.c>

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

//client.cpp

# include "testh.h"

# define TIMEOUT 5

// 뮤텍스 생성 (TCP용)

pthread_mutex_t tcpmutex = PTHREAD_MUTEX_INITIALIZER;

int count = 0;

char sendBuffer[BUFFER_SIZE];

char readBuffer[BUFFER_SIZE];

int receivedBytes;

int target = 0;

int exitNum = 0;

int tmpID;

int tmpIDnum = 0;

int gPort;

int portarr[5];

int tn = 0;

// tcp 소켓 선언

struct sockaddr_in tcplistenSocket[5], tcpconnectSocket[5];

int tcplistenFD[5], tcpconnectFD[5];

struct Table{

int id;

int fd[5];

};

struct Table table[5];

// Alarm에서도 소켓 및 FD가 이용되기에 전역 변수화

struct sockaddr_in listenSocket;

int connectFD;

void alarmHandlerSI(int sigSI)

{

if(exitNum == 1)

{

printf("\nGood Bye!\n");

write(tcpconnectFD[tmpID], "sub logout now",(int)('l'+20));

sendto(connectFD, "exit", (int)(tmpID+'e'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

close(connectFD);

exit(0);

}

if(exitNum == 0)

{

puts("\n[Warning] 'Ctrl + C' key is exit key.");

puts("If you want exit, please try 'Ctrl + C' again");

exitNum = 1;

}

return;

}

void alarmHandler(int sig)

{

count++;

if(count == TIMEOUT)

{

printf("Time Out\n");

exit(0);

}

// 받은 값이 없으면 alarm 시그널 호출

if(receivedBytes == 0)

{

alarm(1);

}

else

{

//응답을 받았으면 alarm 시그널을 취소.

//alarm(0);

count = 0;

receivedBytes = 0;

}

}

void *threadfunc(void *argumentPointer)

{

int connectFD = (*(int*)argumentPointer);

pthread_detach(pthread_self());

while(1)

{

int i;

receivedBytes = read(connectFD, readBuffer, BUFFER_SIZE);

readBuffer[receivedBytes] = '\0';

printf("\n[%d] sub에게 받은 말 :: %s\n",connectFD, readBuffer);

// Subscriber이 종료하였을 경우

if(receivedBytes == (int)('l'+20))

break;

for(i = 0; i < 5; i ++)

{

if(tcpconnectFD[i] != 0 )

{

// pub에게 sub가 입장함을 알려준다.

//tcpconnectSocket[i].sin_port = htons(portarr[i]);

write(connectFD, "Hello subscriber~",30);

}

}

}

close(connectFD);

return NULL;

}

void *threadfuncS(void *ID)

{

pthread_detach(pthread_self());

tmpID = *(int*)ID;

tcpconnectSocket[tmpID].sin_family = AF_INET;

inet_aton("127.0.0.1", (struct in_addr*)&tcpconnectSocket[tmpID].sin_addr.s_addr);

tcpconnectSocket[tmpID].sin_port = htons(gPort);

tcpconnectFD[tmpID] = socket(AF_INET, SOCK_STREAM, 0);

if (connect(tcpconnectFD[tmpID], (struct sockaddr*) &tcpconnectSocket[tmpID], sizeof(tcpconnectSocket[tmpID])) == -1)

{

printf("Can not connect.\n");

sendto(connectFD, "Can not connect", (int)(50+'e'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

}

// 연결 완료. sub에서 pub과 할일을 작성

sendto(connectFD, "Connect Complete", (int)(50+'s'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

//sendto(connectFD, "tmp", 'p', 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

while(1)

{

receivedBytes = read(tcpconnectFD[tmpID], readBuffer, BUFFER_SIZE);

readBuffer[receivedBytes] = '\0';

printf("\npub에게 받은 말 :: %s\n",readBuffer);

printf("\npub에게 보낼 말 :: "); fgets(sendBuffer, BUFFER_SIZE, stdin);

sendBuffer[strlen(sendBuffer)] = '\0';

write(tcpconnectFD[tmpID], sendBuffer,strlen(sendBuffer));

}

close(connectFD);

return;

}

int main(int argc, char** argv)

{

signal(SIGINT, alarmHandlerSI);

// alarm 세팅

struct sigaction alarmAction;

alarmAction.sa_handler = alarmHandler;

alarmAction.sa_flags = 0;

sigaction(SIGALRM, &alarmAction, NULL);

// alarm 시그널 설치

signal(SIGALRM, alarmHandler);

pthread_t threadID[5];

int i;

int state = 0;

int listen_addr_size = sizeof(listenSocket);

memset(&listenSocket, 0, sizeof(listenSocket));

listenSocket.sin_family = AF_INET;

inet_aton("127.0.0.1", (struct in_addr*)&listenSocket.sin_addr.s_addr);

listenSocket.sin_port = htons(PORT);

connectFD = socket(AF_INET, SOCK_DGRAM, 0);

// 알람을 통해 서버의 열림 닫힘 확인

alarm(1);

sprintf(sendBuffer,"%s %s %s %s%c",argv[1],argv[2],argv[3],argv[4],'\0');

sendto(connectFD, sendBuffer, strlen(sendBuffer), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

receivedBytes = recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

readBuffer[receivedBytes] = '\0';

fputs(readBuffer, stdout);

printf("\n");

fflush(stdout);

// recv를 했다는 의미는 서버가열렸다는 의미

alarm(0);

// 에러코드 수신 시 종료

if(receivedBytes == (30+'e') || receivedBytes == (40+'e'))

return 0;

int flag = 0;

int cnt = 0; // sub의 개수

if((int)(0+20+'z') <= receivedBytes && receivedBytes <= (int)(5+20+'z'))

{

printf("Now, you can Publish and Subscribe\n");

//alarm(0);

flag = 3;

fflush(stdout);

//////////

tmpID = receivedBytes - (int)(20+'z');

printf("내 id :: %d\n",tmpID);

table[tmpID].id = tmpID;

tcplistenSocket[tmpID].sin_family = AF_INET;

tcplistenSocket[tmpID].sin_addr.s_addr = htonl(INADDR_ANY);

tcplistenSocket[tmpID].sin_port = htons(PORT+tmpID+1);

tcplistenFD[tmpID] = socket(AF_INET, SOCK_STREAM, 0);

receivedBytes = recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

//sendto(connectFD, "tmp", 5, 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

gPort = receivedBytes+PORT;

// sub할 수 있는 스레드 생성

pthread_mutex_lock(&tcpmutex);

pthread_create(&threadID[tmpID], NULL, threadfuncS, (void *)&tmpID);

pthread_mutex_unlock(&tcpmutex);

if (bind(tcplistenFD[tmpID], (struct sockaddr *) &tcplistenSocket[tmpID], sizeof(tcplistenSocket[tmpID])) == -1) {

printf("Can not bind.\n");

sendto(connectFD, "Can not connect", (int)(50+'e'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

}

if (listen(tcplistenFD[tmpID], LISTEN_QUEUE_SIZE) == -1) {

printf("Listen fail.\n");

return -1;

}

printf("connectFD :: %d\n",connectFD);

while(1)

{

// accept 하는 과정

receivedBytes = recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

if(receivedBytes == (int)('p'))

{

printf("waiting my sub connection...\n");

int connectSocketLength = sizeof(tcpconnectSocket[tmpIDnum]);

while((tcpconnectFD[tmpIDnum] = accept(tcplistenFD[tmpID], (struct sockaddr*)&tcpconnectSocket[tmpID], (socklen_t *)&connectSocketLength)) >= 0)

{

printf("TCP connect complete\n");

if(tn < 5)

{

portarr[tn] = tcpconnectSocket[tmpID].sin_port;

tn++;

}

// pub에서 sub와 할 일을 작성

write(tcpconnectFD[tmpIDnum], "Complete connect to Publisher!", 40);

pthread_mutex_lock(&tcpmutex);

pthread_create(&threadID[tmpIDnum], NULL, threadfunc, (void *)&tcpconnectFD[tmpIDnum]);        tmpIDnum++;

pthread_mutex_unlock(&tcpmutex);

}

}

}

////////

}

if((int)(0+15+'p') <= receivedBytes && receivedBytes <= (int)(5+15+'p'))

{

printf("Now, you can Publish\n");

flag = 2;

fflush(stdout);

tmpID = receivedBytes - (int)(15+'p');

printf("내 id :: %d\n",tmpID);

table[tmpID].id = tmpID;

tcplistenSocket[tmpID].sin_family = AF_INET;

tcplistenSocket[tmpID].sin_addr.s_addr = htonl(INADDR_ANY);

tcplistenSocket[tmpID].sin_port = htons(PORT+tmpID+1);

tcplistenFD[tmpID] = socket(AF_INET, SOCK_STREAM, 0);

if (bind(tcplistenFD[tmpID], (struct sockaddr *) &tcplistenSocket[tmpID], sizeof(tcplistenSocket[tmpID])) == -1) {

printf("Can not bind.\n");

sendto(connectFD, "Can not connect", (int)(50+'e'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

}

if (listen(tcplistenFD[tmpID], LISTEN_QUEUE_SIZE) == -1) {

printf("Listen fail.\n");

return -1;

}

printf("connectFD :: %d\n",connectFD);

while(1)

{

// accept 하는 과정

receivedBytes = recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

if(receivedBytes == (int)('p'))

{

printf("waiting my sub connection...\n");

int connectSocketLength = sizeof(tcpconnectSocket[tmpIDnum]);

while((tcpconnectFD[tmpIDnum] = accept(tcplistenFD[tmpID], (struct sockaddr*)&tcpconnectSocket[tmpID], (socklen_t *)&connectSocketLength)) >= 0)

{

printf("TCP connect complete\n");

if(tn < 5)

{

portarr[tn] = tcpconnectSocket[tmpID].sin_port;

tn++;

}

// pub에서 sub와 할 일을 작성

write(tcpconnectFD[tmpIDnum], "Complete connect to Publisher!", 40);

pthread_mutex_lock(&tcpmutex);

pthread_create(&threadID[tmpIDnum], NULL, threadfunc, (void *)&tcpconnectFD[tmpIDnum]);

tmpIDnum++;

pthread_mutex_unlock(&tcpmutex);

}

}

}

}

else if((int)(0+20+'s') <= receivedBytes && receivedBytes <= (int)(5+20+'s'))

{

printf("Now, you can Subscribe\n");

flag  = 1;

tmpID = receivedBytes - (int)(20+'s');

int tmpPort =recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

readBuffer[tmpPort] = '\0';

//printf("받은 포트 :: %s\n",readBuffer);

tmpPort = atoi(readBuffer) + PORT + 1;

//tcplistenSocket[tmpID].sin_port = htons(tmpPort);

printf("내 id :: %d\n",tmpID);

tcpconnectSocket[tmpID].sin_family = AF_INET;

inet_aton("127.0.0.1", (struct in_addr*)&tcpconnectSocket[tmpID].sin_addr.s_addr);

tcpconnectSocket[tmpID].sin_port = htons(tmpPort);

tcpconnectFD[tmpID] = socket(AF_INET, SOCK_STREAM, 0);

if (connect(tcpconnectFD[tmpID], (struct sockaddr*) &tcpconnectSocket[tmpID], sizeof(tcpconnectSocket[tmpID])) == -1)

{

printf("Can not connect.\n");

sendto(connectFD, "Can not connect", (int)(50+'e'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

return -1;

}

// 연결 완료. sub에서 pub과 할일을 작성

sendto(connectFD, "Connect Complete", (int)(50+'s'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

while(1)

{

receivedBytes = read(tcpconnectFD[tmpID], readBuffer, BUFFER_SIZE);

readBuffer[receivedBytes] = '\0';

printf("\npub에게 받은 말 :: %s\n",readBuffer);

printf("\npub에게 보낼 말 :: "); fgets(sendBuffer, BUFFER_SIZE, stdin);

sendBuffer[strlen(sendBuffer)] = '\0';

write(tcpconnectFD[tmpID], sendBuffer,strlen(sendBuffer));

}

}

pthread_mutex_destroy(&tcpmutex);

//close(connectFD);

return 0;

}

Colored by Color Scripter

cs

1번 결과화면

동영상으로 촬영한 화면 :: https://www.youtube.com/watch?v=J7Ie2wzOEMQ&feature=youtu.be

(0) 컴파일 방법

(1) 처음 Broker을 실행한 화면

(2) 잘못된 argument를 보낼 때

(3) Client가 Publisher로 들어오는 것을 희망하고, room1로 name 설정 및 테이블 갱신

(4) 'Ctrl + c'를 이용하면 종료 직전 경고를 준다.

(5) 'Ctrl + c'를 한번 더 누른 경우, 테이블 갱신 및 종료

(6) 다시 room1을 Publish하고, Subscriber로 하나의 클라이언트가 입장 및 TCP 연결에 성공하여 대화하는 내용

(7) -p name -s name을 통해 Publish 및 Subscribe를 동시에 하고, TCP로 room1과 연결되어 행동하는 모습

(8) 또 다른 client가 Subscriber이 되어 room2에 접근하는 장면

(9) 재사용을 위해 종료 시 테이블이 지속적으로 갱신됨을 나타내는 장면

이때 sub이 종료할 때 마다 pub에 logout이라고 출력된다.

(10) Server(Broker)가 열리지 않았을 시 접속하면 Alarm에 의해 Time Out.

2번 요구 조건 분석(문제 정의)

2. 1번의 시스템을 이용, 아래 요구사항을 만족하는 응용프로그램을 제작하시오. - 2개의 일반 노드를 제작

- 한 노드는 키보드에서 화살표를 입력받아 그 신호를 “command”라는 topic으로

publish함

- 다른 한 노드는 “command” topic을 subscribe하여 화면속의 한 물체를 이동시킴

- 이 노드는 “command” topic의 메시지를 subscribe하지 않더라도, 마치 바람이 불어

물체가 밀리듯이 1초마다 물체를 한쪽으로 이동시킴

- 주의: “command” topic의 메시지를 subscribe하여도 타이머가 reset되어서는 안 됨, 즉 사용자의 키보드 입력과 상관없이 매 1초마다 물체가 밀려나가야 함

- Alarm signal을 사용해도 되고 thread를 사용해도 됨

2번 요구사항 알고리즘(문제 분석)

1번 요구사항과 유사한 내용(아래 2번의 응용 내용 기재)

가장 처음에는 클라이언트가 브로커에게 최대 4개의 argument를 string으로 변환 후 보낼 수 있도록 하여 브로커가 string을 확인 후 각 argument에 어떤 값들이 들어왔는지 확인하는 방식을 이용한다.

위의 그림처럼 -p name인지, -s name인지, -p name -s name인지 파악하여 각 flag로 나누고

해당하는 flag의 조건문에서 Client가 어떤 방식을 취할지 Broker가 설정해준다.

flag가 1일 때(Publisher)

Broker 입장

Client이 가지는 제한사항이 없으니 보내주고, Broker Table에 업데이트한다.

Client 입장

브로커에서 보내준 size값을 이용하여(15+‘p’라는 값을 암호 키처럼 보내준다.) 자신의 아이디와 포트 번호를 알도록 해주고 TCP를 이용하여 bind, listen을 통해 accept를 할 준비를 한다. 이때 accept과정은 Broker이 Subscriber이 들어온 경우 while문에서 ‘p’라는 암호 키를 보내주면 accept를 준비하도록 설계되어있다. accept를 성공하면 mutex_lock을 통해 스레드를 생성하고 mutex_unlock를 해준다. 이때 스레드(threadfunc)에서는 맵을 생성 후 sub에게 방향키 메시지를 받게 되면 그에 맞게 poll을 통해 1초에 한번씩 움직이게 된다. 그리고 클라이언트가 접속하여 TCP 통신에 성공하에 된다면, while문 getch()를 통해 pub에게 'o'가 움직일 방향을 줄 수 있고, pub과 연결이 되었음을 확인하기 위해서는 pub의 맵에 'o'가 움직이는 것을 통해 확인 할 수 있다.

flag가 2일 때(Subscriber)

Broker 입장

Subscriber일 때는 Publisher가 존재하는지 확인해야한다.(Name matching)

따라서 이름을 비교하는 알고리즘을 이용한다. 이 알고리즘은 for문으로 strcmp를 이용하여 Broker Table에서 찾아내는 방식을 이용한다.

방이 존재하지 않는다면 flag가 -2로 변하고, “ERROR :: Room is not exist”이라는 메시지를 클라이언트에 보내주며 클라이언트를 종료 시킨다.

방이 존재한다면 클라이언트에게 접속 시도를 할 수 있도록 해주고, Subscriber는 Publisher에게 접속을 시도하게 되고, Broker Table에 업데이트한다.

Client 입장

브로커에서 보내준 size값을 이용하여(20+'s'라는 값을 암호 키처럼 보내준다.) 자신의 아이디와 포트 번호를 알도록 해주고 connect를 준비한다. TCP를 통해 connect를 성공하면 while문 getch()를 통해 pub에게 'o'가 움직일 방향을 줄 수 있고, pub과 연결이 되었음을 확인하기 위해서는 pub의 맵에 'o'가 움직이는 것을 통해 확인 할 수 있다.

flag가 3일 때(Publisher and Subscriber)

Broker 입장

flag1과 flag2가 가지는 조건을 이용한다. 따라서 for문으로 strcmp를 통해 Broker Table에서 Publisher이 있는지 찾아내는 알고리즘을 동일하게 이용하고, Broker Table에 업데이트한다. 이때도 마찬가지로 방이 존재하지 않을 경우 flag가 -3이 되어 “ERROR :: Room is not exist”이라는 메시지를 클라이언트에 보내주며 클라이언트를 종료 시킨다.

Client 입장

브로커에서 보내준 size값을 이용하여(20+'z'라는 값을 암호 키처럼 보내준다.) 자신의 아이디와 포트 번호를 알도록 해주고 mutex_lock를 통해 threadfuncS라는 스레드를 생성한다. 이 스레드는 Subscriber 역할 및 Publisher 역할을 할 수 있는 스레드로써 TCP를 통해 Publisher 및 Subscriber와 연결을 한다. 그리고나서 while문을 통해 Publisher에게 메시를 보내고, 정상적으로 통신이 되었다면 Publisher에게 getch()를 통해 방향키 값을 줄 수 있다. 그리고 메인 스레드에서는 bind, listen을 통해 accept를 준비하고 마찬가지로 'p'라는 암호 키를 받으면 accept를 할 수 있도록 하여 publish와 subscribe를 동시에 할 수 있도록 하였다.

2번 응용 내용

1. getch()의 이용

getch()의 함수는 다음과 같다.

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15	int getch(void) { int ch; struct termios buf; struct termios save; tcgetattr(0, &save); buf = save; buf.c_lflag &= ~(ICANON | ECHO); buf.c_cc[VMIN] = 1; buf.c_cc[VTIME] = 0; tcsetattr(0, TCSAFLUSH, &buf); ch = getchar(); tcsetattr(0, TCSAFLUSH, &save); return ch; }	cs

리눅스에서는 kbhit같은 내용이 없으므로 getch라는 새로운 함수를 구현하여 이용한다.

이 함수는 키가 눌러짐과 동시에 반응을 해주는 역할을 한다. 즉, s를 누르면 return ch를 통해 s가 바로 전달된다. 이 방식을 이용하여 방향키의 모션을 만들어 줄 수 있다.

2. gotoxy()의 이용

콘솔 그래픽을 이용하기 위해서 필수적인 함수이다.

gotoxy함수를 통해 원하는 좌표 점에 출력을 할 수 있고, 이를 통해 맵을 그리고, 출력문을 만들었다. 스레드를 이용함에 있어 항상 조심해야 할 점은 커서는 하나이기에 스레드와 메인 스레드 사이 printf가 동시에 있을 경우 출력이 엉망으로 나올 수 있으니 mutex혹은 메인 스레드 혹은 자식 스레드 한군데 에서만 printf를 이용할 수 있는 방식을 고려해야 한다.

1 2 3 4	void gotoxy(int x,int y) { printf("%c[%d;%df",0x1B,y,x); }	cs

3. poll()의 이용

실습 시간에 배운 것을 토대로 poll을 적용시켰다. 그리고 poll에서 1초 동안 대기하며 타임아웃이 된다면 방향키에 맞는 방향으로 움직이도록 하였고, posX, posY가 맵에 부딪히지 않도록 경계를 설정해두었다. 이때 1초 내에 값이 들어온다면 clock의 start와 end를 구하여 sum = start - end로 지정하고 poll에서 1000-sum을 통해 최대한 1초에 맞도록 하였다. 많은 스레드와 프로세스가 돌고있기에 1초가 정확하게는 닿지 못하지만 최대한 근사치로 수렴하도록 동기화 시키는 알고리즘을 구상하였다.

2번 소스 코드

<Test.h>

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15	//testh.h #include <pthread.h> #include <semaphore.h> #include <stdio.h> #include <string.h> #include <stdlib.h> #include <netinet/in.h> #include <sys/socket.h> #include <arpa/inet.h> #include <unistd.h> #include <signal.h> #define PORT 20162 #define BUFFER_SIZE 300 #define LISTEN_QUEUE_SIZE 5 #define TIMEOUT 5	cs

<Server.c>

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

//server.cpp

# include "testh.h"

// 뮤텍스 생성 및 컨디션 생성(UDP용)

pthread_mutex_t udpmutex = PTHREAD_MUTEX_INITIALIZER;

pthread_cond_t udpcond = PTHREAD_COND_INITIALIZER;

int total = 0;

int count = 0;

int empty[5];

int emptyPos = 0;

int emptyFlag = 0;

int errorFlag = 0;

int now;

char readBuff[BUFFER_SIZE];

char sendBuff[BUFFER_SIZE];

int listenFD;

int connectFD;

int target; // 몇 번 방인지 확인하는 변수

int receivedBytes;

struct sockaddr_in listenSocket, connectSocket;

int connect_addr_size = sizeof(connectSocket);

// 알람을 통해 테이블을 갱신한다

void alarmHandler(int sig)

{

count++;

if(count == TIMEOUT)

{

printf("Time Out\n");

exit(0);

}

// 받은 값이 없으면 alarm 시그널 호출

if(receivedBytes == 0)

{

alarm(1);

}

else

{

//응답을 받았으면 alarm 시그널을 취소.

alarm(0);

count = 0;

receivedBytes = 0;

}

}

// 테이블 생성

struct Table

{

char createroom[100];

char joinroom[100];

//int connectPort[5];

int pub;

int sub;

int id;

int fd;

int port;

};

struct Table table[5]; // 최대 5개 서버 가능

void sendport(int port)

{

// port를 char형으로 변환하여 전송하는 알고리즘

char bufftmp[100];

int portTmp = port;

int i;

for(i = 0; ;i ++)

{

if(portTmp / 10 != 0)

{

bufftmp[i] = (portTmp % 10) + 48; // ASCII

portTmp /= 10;

}

else

break;

}

bufftmp[i] = (portTmp % 10) + 48; // 마지막 값

bufftmp[i+1] = '\0';

int j = strlen(bufftmp);

for(i = 0; j > 0 ; i++)

{

readBuff[j-1] = bufftmp[i];

j--;

}

readBuff[i] = '\0';

// publisher에게는 subscriber의 port를 준다.

sendto(listenFD, readBuff, strlen(readBuff),0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

}

int getinit()

{

int i = 0;

int j = 0;

int flag = 0;

// 테이블에 빈자리가 있다면 빈자리를 채우는 알고리즘

if(emptyFlag == 1)

{

emptyFlag = 0;

emptyPos--;

}

else if(emptyPos != 0)

{

now = empty[emptyPos-1];

//emptyPos--;

emptyFlag = 1;

i = 5;

}

else if(emptyPos == 0 && emptyFlag == 0)

{

int tmpi;

for(tmpi = 0; tmpi < 5; tmpi++)

{

if(table[tmpi].fd == 0)

{

now = tmpi;

//errorFlag = 0;

break;

}

}

}

table[now].createroom[0] = '\0';

table[now].joinroom[0] = '\0';

table[now].pub = 0;

table[now].sub = 0;

table[now].id = now;

table[now].fd = 0;

table[now].port = 0;

int tmpi;

for(tmpi = 0; tmpi < 5; tmpi++)

{

if(table[tmpi].fd == 0)

{

errorFlag == 1 ? (now = tmpi) : (now = tmpi);

errorFlag = 0;

break;

}

}

// 시작이 -p or -s가 아닐 때

if(!((readBuff[0] == '-' && readBuff[1] == 's' && readBuff[2] == ' ') || (readBuff[0] == '-' && readBuff[1] == 'p' && readBuff[2] == ' ')))

{

sendto(listenFD, "ERROR :: Usage :: '-p name -s name' or '-p name' or '-s name'", 20+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket)+30);

return -1;

}

// 시작이 -p 네임 일 때

if(readBuff[0] == '-' && readBuff[1] == 'p' && readBuff[2] == ' ' && !strstr(readBuff,"-s"))

{

table[now].pub = 1;

table[now].fd = listenFD;

table[now].port = connectSocket.sin_port;

i = 3;

for(i; readBuff[i] != ' '; i++,j++)

table[now].createroom[j] = readBuff[i];

table[now].createroom[j] = '\0';

i++;

flag = 1;

}

j = 0;

// 시작이 -s 네임 일 때

if(readBuff[0] == '-' && readBuff[1] == 's' && readBuff[2] == ' ')

{

table[now].sub = 1;

table[now].fd = listenFD;

table[now].port = connectSocket.sin_port;

i += 3;

for(i; readBuff[i] != ' '; i++,j++)

table[now].joinroom[j] = readBuff[i];

table[now].joinroom[j] = '\0';

i++;

flag = 2;

}

// 시작이 -p 네임 -s 네임 일 때

//if(readBuff[i] == '-'  readBuff[i+1] == 'p'  readBuff[2] == ' ')

if(strstr(readBuff,"-p") && strstr(readBuff,"-s"))

{

table[now].pub = 1;

table[now].sub = 1;

table[now].fd = listenFD;

table[now].port = connectSocket.sin_port;

for(i = 3; readBuff[i] != ' '; i++,j++)

table[now].createroom[j] = readBuff[i];

table[now].createroom[j] = '\0';

i += 4;

j = 0;

for(i; i < strlen(readBuff); i++,j++)

{

table[now].joinroom[j] = readBuff[i];

}

table[now].joinroom[j] = '\0';

flag = 3;

i = 5;

}

// -p 네임 -'이상한 문자' 일 때

else if(readBuff[i] == '-' && readBuff[i+1] != 's' && readBuff[2] == ' ')

{

sendto(listenFD, "ERROR :: Usage :: '-p name -s name' or '-p name' or '-s name'", 20+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

return -1;

}

// 현재 들어온 값 보여준다.

printf("id :: %d\npub :: %d\nsub :: %d\nfd :: %2d\nport :: %2d\nname :: %s\nsname :: %s\n",table[now].id,table[now].pub,table[now].sub,table[i].fd, table[i].port,table[now].createroom,table[now].joinroom);

// 갱신된 테이블 값 보여준다.

printf("잘못된 입력 값도 받아 오는 [Table]\n");

for(i = 0; i < 5; i ++)

{

printf("id :: %d pub :: %d sub :: %d fd :: %2d port :: %5d pname :: %5s sname :: %5s\n",table[i].id,table[i].pub,table[i].sub,table[i].fd, table[i].port,table[i].createroom,table[i].joinroom);

}

// 플래그 상태(1 :: pub, 2:: sub, 3 :: pub  sub)

// 상태에 따라 pub, sub, pub&sub인지 알려준다.

printf("flag :: %d\n",flag);

if(flag == 1)

{

sendto(listenFD, "Broker Connect Completely", table[now].id + 15+'p', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

}

else if(flag == 2)

{

flag = -2;

for(i = 0; i < 5; i ++)

{

// 방이 생성되어있는지 확인한다.

if(strcmp(table[i].createroom,table[now].joinroom) == 0)

{

/*

알고리즘 ::

우선 sub에게 방이 존재하니 커넥트를 완료했음을 알린다.

그리고 pub에게 sub가 입장함을 알린다.

그 후, sub가 fd를 전달해주면 브로커는 sub의 fd를

pub에게 전달해주고, pub는 그것을 수신하게되면

pub의 fd를 다시 브로커에게 전달해준다.

이것을 sub에게 마지막으로 전달해주면 완료.

*/

sendto(listenFD, "Broker Connect Completely", table[now].id + 20+'s', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

sendport(i);

flag = 2;

//                int subFD = recvfrom(listenFD, readBuff, BUFFER_SIZE, 0, (struct sockaddr*)&connectSocket, &connect_addr_size);

int tmpPort = connectSocket.sin_port; // sub의 포트

// pub에게 sub가 입장함을 알려준다.

connectSocket.sin_port = table[i].port;

sendto(listenFD, "Try to accept", 'p', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

receivedBytes = recvfrom(listenFD, readBuff, BUFFER_SIZE, 0, (struct sockaddr*)&connectSocket, &connect_addr_size);

if(receivedBytes == (int)(50 + 'e'))

{

table[now].createroom[0] = '\0';

table[now].joinroom[0] = '\0';

table[now].pub = 0;

table[now].sub = 0;

table[now].id = now;

table[now].fd = 0;

table[now].port = 0;

//total --;

return -1;

}

break;

}

}

// 방이 없을경우 40 + 'e' 아스키 코드 전송

if(flag == -2)

{

errorFlag = 1;

sendto(listenFD, "ERROR :: Room is not exist", 40+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

table[now].createroom[0] = '\0';

table[now].joinroom[0] = '\0';

table[now].pub = 0;

table[now].sub = 0;

table[now].id = 0;

table[now].fd = 0;

table[now].port = 0;

return -1;

}

}

else if(flag == 3)

{

flag = -3;

for(i = 0; i < 5; i ++)

{

if(i != now && strcmp(table[i].createroom,table[now].joinroom) == 0)

{

sendto(listenFD, "Broker Connect Completely", table[now].id + 20+'z', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

sendto(listenFD, "p",  table[i].id+1,0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

int tmpPort = connectSocket.sin_port; // sub의 포트

// pub에게 sub가 입장함을 알려준다.

connectSocket.sin_port = table[i].port;

sendto(listenFD, "Try to accept", 'p', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

flag = 3;

break;

}

}

// 방이 없을경우 40 + 'e' 아스키 코드 전송

if(flag == -3)

{

errorFlag = 1;

sendto(listenFD, "ERROR :: Room is not exist", 40+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

table[now].createroom[0] = '\0';

table[now].joinroom[0] = '\0';

table[now].pub = 0;

table[now].sub = 0;

table[now].id = 0;

table[now].fd = 0;

table[now].port = 0;

return -1;

}

}

return 1;

}

int main()

{

int i;

int flag = 0;

pthread_t threadID[5];

memset(&listenSocket, 0, sizeof(listenSocket));

memset(&listenSocket, 0, sizeof(connectSocket));

listenSocket.sin_family = AF_INET;

listenSocket.sin_addr.s_addr = htonl(INADDR_ANY);

listenSocket.sin_port = htons(PORT);

// 브로커 서버 소켓 생성(UDP :: SOCK_DGRAM)

if((listenFD = socket(AF_INET, SOCK_DGRAM, 0)) == -1)

{

printf("Server :: Can not Open Socket\n");

exit(0);

}

if(bind(listenFD, (struct sockaddr *) &listenSocket, sizeof(listenSocket)) == -1)

{

printf("Can not bind\n");

return -1;

}

printf("Waiting for clients...\n");

while(1)

{

receivedBytes = recvfrom(listenFD, readBuff, BUFFER_SIZE, 0, (struct sockaddr*)&connectSocket, &connect_addr_size);

readBuff[receivedBytes] = '\0';

printf("--------------------\n");

//printf("::: %s\n", readBuff);

// 종료 메시지를 보내온다면 테이블 정리를 한다.

if(0 + 'e' <= receivedBytes && receivedBytes <= 5 + 'e')

{

total--;

int tmpID = receivedBytes - 'e';

printf("지금 종료한 id :: %d\n",tmpID);

// 테이블의 빈자리를 알게 해준다.

empty[emptyPos] = tmpID;

emptyPos++;

table[tmpID].createroom[0] = '\0';

table[tmpID].joinroom[0] = '\0';

table[tmpID].pub = 0;

table[tmpID].sub = 0;

table[tmpID].id = 0;

table[tmpID].fd = 0;

table[tmpID].port = 0;

}

if(total < 5)

flag = getinit(); // -p 네임 -s 네임 확인

else

sendto(listenFD, "ERROR :: Client is full", 30+'e', 0, (struct sockaddr*)&connectSocket, sizeof(connectSocket));

flag == 1 ? total++ : 0;

flag = 0;

printf("total :: %d\n",total);

// 갱신된 테이블 값 보여준다.

printf("오류를 처리한 최종 [Table]\n");

for(i = 0; i < 5; i ++)

{

printf("id :: %d pub :: %d sub :: %d fd :: %2d port :: %5d pname :: %5s sname :: %5s\n",table[i].id,table[i].pub,table[i].sub,table[i].fd, table[i].port,table[i].createroom,table[i].joinroom);

}

printf("\n");

fflush(stdout);

}

}

Colored by Color Scripter

cs

<Client.c>

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

//client.cpp

#include "testh.h"

#include <termio.h>

#include <unistd.h>

#include <poll.h>

#include <time.h>

# define TIMEOUT 5

// 뮤텍스 생성 (TCP용)

pthread_mutex_t tcpmutex = PTHREAD_MUTEX_INITIALIZER;

int clearflag = 0;

char mkey = '-';

int posX,posY;

int threadflag = 0;

double start,end;

int count = 0;

char sendBuffer[BUFFER_SIZE];

char readBuffer[BUFFER_SIZE];

int receivedBytes;

int target = 0;

int exitNum = 0;

int tmpID;

int tmpIDnum = 0;

int gPort;

// tcp 소켓 선언

struct sockaddr_in tcplistenSocket[5], tcpconnectSocket[5];

int tcplistenFD[5], tcpconnectFD[5];

struct Table{

int id;

int fd[5];

};

struct Table table[5];

// Alarm에서도 소켓 및 FD가 이용되기에 전역 변수화

struct sockaddr_in listenSocket;

int connectFD;

void alarmHandlerSI(int sigSI)

{

if(exitNum == 1)

{

printf("\nGood Bye!\n");

write(tcpconnectFD[tmpID], "sub logout now",(int)('l'+20));

sendto(connectFD, "exit", (int)(tmpID+'e'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

close(connectFD);

exit(0);

}

if(exitNum == 0)

{

puts("\n[Warning] 'Ctrl + C' key is exit key.");

puts("If you want exit, please try 'Ctrl + C' again");

exitNum = 1;

}

return;

}

void alarmHandler(int sig)

{

count++;

if(count == TIMEOUT)

{

printf("Time Out\n");

exit(0);

}

// 받은 값이 없으면 alarm 시그널 호출

if(receivedBytes == 0)

{

alarm(1);

}

else

{

//응답을 받았으면 alarm 시그널을 취소.

//alarm(0);

count = 0;

receivedBytes = 0;

}

}

void gotoxy(int x,int y)

{

printf("%c[%d;%df",0x1B,y,x);

}

int getch(void)

{

int ch;

struct termios buf;

struct termios save;

tcgetattr(0, &save);

buf = save;

buf.c_lflag &= ~(ICANON | ECHO);

buf.c_cc[VMIN] = 1;

buf.c_cc[VTIME] = 0;

tcsetattr(0, TCSAFLUSH, &buf);

ch = getchar();

tcsetattr(0, TCSAFLUSH, &save);

return ch;

}

void *threadfunc(void *argumentPointer)

{

int connectFD = (*(int*)argumentPointer);

int tflag = threadflag;

pthread_detach(pthread_self());

int l;

if(clearflag == 0)

{

system("clear"); //clears the screen

clearflag = 1;

}

if(tflag == 1)

{

for(l = 1; l < 10 ; l++)

{gotoxy(1,l);printf("*                   *");}

for(l = 1; l <= 20; l++)

{gotoxy(l,1); printf("*");}

for(l = 1; l <= 21; l++)

{gotoxy(l,10); printf("*");}

gotoxy(10,5); printf("o");

posX = 10;

posY = 5;

}

///////////////

// pollfd 배열 구조체 생성

struct pollfd pollFDs[5];

int i;

pollFDs[0].fd = connectFD;

pollFDs[0].events = POLLIN; // 읽도록 만든다.

pollFDs[0].revents = 0; // 처음에는 0으로 초기화 한다(아직 아무 일도 일어나지 않았으니)

for (i = 1; i < 5; i++)

pollFDs[i].fd = -1;

while (1)

{

int sum = 0;

// 1초 동기화 알고리즘

end = clock();

if(start != 0)

sum = start - end;

int result = poll(pollFDs, 5, 1000 - sum); // 1초 대기

if (result == 0)

{

gotoxy(0,13);printf("현재 x :: %d  y :: %d \n",posX,posY);

// 위로

if(tflag == 1)

{

if((mkey == 'w' || mkey =='W') && (2 < posY  && posY <= 9))

{

gotoxy(posX,posY); printf(" ");

gotoxy(posX,posY-1); printf("o");

posX = posX;

posY = posY - 1;

}

else if((mkey == 's' || mkey =='S') && (1 <= posY  && posY < 9))

{

gotoxy(posX,posY); printf(" ");

gotoxy(posX,posY+1); printf("o");

posX = posX;

posY = posY + 1;

}

else if((mkey == 'a' || mkey =='A') && (2 < posX  && posX <= 20))

{

gotoxy(posX,posY); printf(" ");

gotoxy(posX - 1,posY); printf("o");

posX = posX - 1;

posY = posY;

}

else if((mkey == 'd' || mkey =='D') && (2 <= posX  && posX < 20))

{

gotoxy(posX,posY); printf(" ");

gotoxy(posX + 1,posY); printf("o");

posX = posX + 1;

posY = posY;

}

}

}

else if (result > 0)

{

start = clock();

if (pollFDs[0].revents == POLLIN)

{

int i;

receivedBytes = read(connectFD, readBuffer, BUFFER_SIZE);

readBuffer[receivedBytes] = '\0';

gotoxy(1,12);printf("Subscriber key :: %s   \n", readBuffer);

if(readBuffer[0] == 'w' || readBuffer[0] == 's' || readBuffer[0] == 'a' || readBuffer[0] == 'd')

mkey = readBuffer[0];

// Subscriber이 종료하였을 경우

if(receivedBytes == (int)('l'+20))

{

break;

}

}

}

}

close(connectFD);

return NULL;

}

void *threadfuncS(void *ID)

{

pthread_detach(pthread_self());

tmpID = *(int*)ID;

tcpconnectSocket[tmpID].sin_family = AF_INET;

inet_aton("127.0.0.1", (struct in_addr*)&tcpconnectSocket[tmpID].sin_addr.s_addr);

tcpconnectSocket[tmpID].sin_port = htons(gPort);

tcpconnectFD[tmpID] = socket(AF_INET, SOCK_STREAM, 0);

if (connect(tcpconnectFD[tmpID], (struct sockaddr*) &tcpconnectSocket[tmpID], sizeof(tcpconnectSocket[tmpID])) == -1)

{

printf("Can not connect.\n");

sendto(connectFD, "Can not connect", (int)(50+'e'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

}

// 연결 완료. sub에서 pub과 할일을 작성

sendto(connectFD, "Connect Complete", (int)(50+'s'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

system("clear");

while(1)

{

char ch = getch();

gotoxy(1,11);printf("press key :: %c\n",ch);

sprintf(sendBuffer,"%c",ch);

sendBuffer[strlen(sendBuffer)] = '\0';

write(tcpconnectFD[tmpID], sendBuffer,strlen(sendBuffer));

}

close(connectFD);

return;

}

int main(int argc, char** argv)

{

signal(SIGINT, alarmHandlerSI);

// alarm 세팅

struct sigaction alarmAction;

alarmAction.sa_handler = alarmHandler;

alarmAction.sa_flags = 0;

sigaction(SIGALRM, &alarmAction, NULL);

// alarm 시그널 설치

signal(SIGALRM, alarmHandler);

pthread_t threadID[5];

int i;

int state = 0;

int listen_addr_size = sizeof(listenSocket);

memset(&listenSocket, 0, sizeof(listenSocket));

listenSocket.sin_family = AF_INET;

inet_aton("127.0.0.1", (struct in_addr*)&listenSocket.sin_addr.s_addr);

listenSocket.sin_port = htons(PORT);

connectFD = socket(AF_INET, SOCK_DGRAM, 0);

// 알람을 통해 서버의 열림 닫힘 확인

alarm(1);

sprintf(sendBuffer,"%s %s %s %s%c",argv[1],argv[2],argv[3],argv[4],'\0');

sendto(connectFD, sendBuffer, strlen(sendBuffer), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

receivedBytes = recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

readBuffer[receivedBytes] = '\0';

fputs(readBuffer, stdout);

printf("\n");

fflush(stdout);

// recv를 했다는 의미는 서버가열렸다는 의미

alarm(0);

// 에러코드 수신 시 종료

if(receivedBytes == (30+'e') || receivedBytes == (40+'e'))

return 0;

int flag = 0;

int cnt = 0; // sub의 개수

if((int)(0+20+'z') <= receivedBytes && receivedBytes <= (int)(5+20+'z'))

{

printf("Now, you can Publish and Subscribe\n");

//alarm(0);

flag = 3;

fflush(stdout);

//////////

tmpID = receivedBytes - (int)(20+'z');

printf("내 id :: %d port :: %d\n",tmpID,PORT+tmpID+1);

table[tmpID].id = tmpID;

tcplistenSocket[tmpID].sin_family = AF_INET;

tcplistenSocket[tmpID].sin_addr.s_addr = htonl(INADDR_ANY);

tcplistenSocket[tmpID].sin_port = htons(PORT+tmpID+1);

tcplistenFD[tmpID] = socket(AF_INET, SOCK_STREAM, 0);

receivedBytes = recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

gPort = receivedBytes+PORT;

// sub할 수 있는 스레드 생성

pthread_mutex_lock(&tcpmutex);

pthread_create(&threadID[tmpID], NULL, threadfuncS, (void *)&tmpID);

pthread_mutex_unlock(&tcpmutex);

if (bind(tcplistenFD[tmpID], (struct sockaddr *) &tcplistenSocket[tmpID], sizeof(tcplistenSocket[tmpID])) == -1) {

printf("Can not bind.\n");

return -1;

}

if (listen(tcplistenFD[tmpID], LISTEN_QUEUE_SIZE) == -1) {

printf("Listen fail.\n");

return -1;

}

printf("connectFD :: %d\n",connectFD);

printf("wating my sub connection...\n");

while(1)

{

// accept 하는 과정

receivedBytes =recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

if(receivedBytes == (int)('p'))

{

int connectSocketLength = sizeof(tcpconnectSocket[tmpIDnum]);

while((tcpconnectFD[tmpIDnum] = accept(tcplistenFD[tmpID], (struct sockaddr*)&tcpconnectSocket[tmpID], (socklen_t *)&connectSocketLength)) >= 0)

{

//printf("TCP connect complete\n");

// pub에서 sub와 할 일을 작성

write(tcpconnectFD[tmpIDnum], "Complete connect to Publisher!", 40);

pthread_mutex_lock(&tcpmutex);

pthread_create(&threadID[tmpIDnum], NULL, threadfunc, (void *)&tcpconnectFD[tmpIDnum]);

tmpIDnum++;

threadflag++;

pthread_mutex_unlock(&tcpmutex);

}

}

}

////////

}

if((int)(0+15+'p') <= receivedBytes && receivedBytes <= (int)(5+15+'p'))

{

printf("Now, you can Publish\n");

flag = 2;

fflush(stdout);

tmpID = receivedBytes - (int)(15+'p');

printf("내 id :: %d port :: %d\n",tmpID,PORT+tmpID+1);

table[tmpID].id = tmpID;

tcplistenSocket[tmpID].sin_family = AF_INET;

tcplistenSocket[tmpID].sin_addr.s_addr = htonl(INADDR_ANY);

tcplistenSocket[tmpID].sin_port = htons(PORT+tmpID+1);

tcplistenFD[tmpID] = socket(AF_INET, SOCK_STREAM, 0);

if (bind(tcplistenFD[tmpID], (struct sockaddr *) &tcplistenSocket[tmpID], sizeof(tcplistenSocket[tmpID])) == -1) {

printf("Can not bind.\n");

return -1;

}

if (listen(tcplistenFD[tmpID], LISTEN_QUEUE_SIZE) == -1) {

printf("Listen fail.\n");

return -1;

}

printf("connectFD :: %d\n",connectFD);

printf("wating my sub connection...\n");

while(1)

{

// accept 하는 과정

receivedBytes =recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

if(receivedBytes == (int)('p'))

{

int connectSocketLength = sizeof(tcpconnectSocket[tmpIDnum]);

while((tcpconnectFD[tmpIDnum] = accept(tcplistenFD[tmpID], (struct sockaddr*)&tcpconnectSocket[tmpID], (socklen_t *)&connectSocketLength)) >= 0)

{

//printf("TCP connect complete\n");

// pub에서 sub와 할 일을 작성

write(tcpconnectFD[tmpIDnum], "Complete connect to Publisher!", 40);

pthread_mutex_lock(&tcpmutex);

pthread_create(&threadID[tmpIDnum], NULL, threadfunc, (void *)&tcpconnectFD[tmpIDnum]);

tmpIDnum++;

threadflag++;

pthread_mutex_unlock(&tcpmutex);

}

}

}

}

else if((int)(0+20+'s') <= receivedBytes && receivedBytes <= (int)(5+20+'s'))

{

printf("Now, you can Subscribe\n");

flag  = 1;

tmpID = receivedBytes - (int)(20+'s');

int tmpPort =recvfrom(connectFD, readBuffer, BUFFER_SIZE, 0, (struct sockaddr *)&listenSocket, &listen_addr_size);

readBuffer[tmpPort] = '\0';

//printf("받은 포트 :: %s\n",readBuffer);

tmpPort = atoi(readBuffer) + PORT + 1;

//tcplistenSocket[tmpID].sin_port = htons(tmpPort);

printf("내 id :: %d port :: %d\n",tmpID,tmpPort);

tcpconnectSocket[tmpID].sin_family = AF_INET;

inet_aton("127.0.0.1", (struct in_addr*)&tcpconnectSocket[tmpID].sin_addr.s_addr);

tcpconnectSocket[tmpID].sin_port = htons(tmpPort);

tcpconnectFD[tmpID] = socket(AF_INET, SOCK_STREAM, 0);

if (connect(tcpconnectFD[tmpID], (struct sockaddr*) &tcpconnectSocket[tmpID], sizeof(tcpconnectSocket[tmpID])) == -1)

{

printf("Can not connect.\n");

sendto(connectFD, "Can not connect", (int)(50+'e'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

return -1;

}

// 연결 완료. sub에서 pub과 할일을 작성

sendto(connectFD, "Connect Complete", (int)(50+'s'), 0, (struct sockaddr*)&listenSocket, sizeof(listenSocket));

system("clear");

while(1)

{

char ch = getch();

gotoxy(1,10);printf("press key :: %c\n",ch);

sprintf(sendBuffer,"%c",ch);

sendBuffer[strlen(sendBuffer)] = '\0';

write(tcpconnectFD[tmpID], sendBuffer,strlen(sendBuffer));

}

}

pthread_mutex_destroy(&tcpmutex);

//close(connectFD);

return 0;

}

Colored by Color Scripter

cs

2번 결과화면

동영상으로 촬영한 화면 :: https://www.youtube.com/watch?v=DVWx-Hj_KPc

(0) 컴파일 방법

(1) 처음 Broker을 실행한 화면

(2) 하나의 client가 들어왔을 때 (-p room1)

sub을 기다리고 있는 모습임을 나타낸다.

(3) 하나의 client가 들어왔을 때 (-s room1)

room1의 publisher와 연결되고 sub에서 키를 누르면 움직인다.

(3) 하나의 client가 들어왔을 때 (-p room2 -s room1)

room1의 publisher와 연결되고 sub에서 키를 누르면 움직인다.

또한 다른 sub이 오면 방금 생긴 터미널 또한 맵을 그리게 된다.

(3) 하나의 client가 들어왔을 때 (-s room2)

-p name -s name또한 맵을 그리고 움직임을 알 수 있다.